生产者生产了短信,2个小时后消费者才将短信发送出去,问题出在了哪里
悬赏:20 发布时间:2008-05-13 提问人:lzmhehe (初级程序员)
才发现这里有个问答模块
转发一下
整个算法大致是:
多个线程同时生产短信,放到队列中(存在优先级,且先进先出),
一个消费线程从队列中拿出短信发送出去
目前出现了一个现象
生产者生产了短信,2个小时后消费者才将短信发送出去
而这2个小时的时间里面,又生产的其他短信都比较正常,都能马上发送出去,就这个停在队列里面,像是放假了不起床
并且程序运行了快半年了都没有出现这个毛病
生产者方面
MessageThread run 方法
消费者
队列结构:
问题补充:
你在造轮子啊。。。。。
为什么说我在造轮子呢,有现成的东西可以使用么
========================
按照你的描述 问题在生产者和消费者的可能不太 看看发送模块是否有异样日志
正式环境下面消费者(发送短信)消费一次需要几秒甚至一分多
而我在测试环境下面消费者使用了mock
发送了几万条一点问题都没有
不知道什么原因
另外生产环境下消费者是其他厂商提供的api
该问题已经关闭: 超过15天由系统自动关闭
转发一下
整个算法大致是:
多个线程同时生产短信,放到队列中(存在优先级,且先进先出),
一个消费线程从队列中拿出短信发送出去
目前出现了一个现象
生产者生产了短信,2个小时后消费者才将短信发送出去
而这2个小时的时间里面,又生产的其他短信都比较正常,都能马上发送出去,就这个停在队列里面,像是放假了不起床
并且程序运行了快半年了都没有出现这个毛病
生产者方面
public void sendSms(String xml) throws RemoteException {
log.info("Rmi接口接收到了xml"+xml);
MessageThread messageThread = Factory.getMessageThread();
messageThread.setXml(xml);
VerifyResult verifyResult = messageThread.verifyXml();
if (verifyResult.isValid()||verifyResult.equals(VerifyResult.XMLBEAN_HAS_DUPLE_PHONE)) {
log.info("过滤成功,开始放入发送队列");
new Thread(messageThread).start();
} else {
log.error("短信过滤掉了,原因:"+verifyResult.getMessage());
}
}
MessageThread run 方法
public void run() {
if (verified) {
ShortMessage shortMessage = xmlBean.convert2SM();
shortMessage.setServiceNo(shortMsgServiceNo.getServiceNo(xmlBean));
if (log4j.isDebugEnabled()) {
log4j.debug("shortMessage: 开始放入队列:" + shortMessage);
}
synchronized (queue) {
queue.push(shortMessage);
queue.notifyAll();
}
} else {
throw new ServiceException("请先检验在执行,当前没有检验");
}
}
消费者
public void run() {
stop=false;
if(log.isDebugEnabled()){
log.debug("SmsSendManager start run");
}
while (true){
ShortMessage sm =null;
synchronized (queue) {
while (queue.isEmpty()&&!stop) {
try {
queue.wait();
} catch (InterruptedException e) {
e.printStackTrace();
log.error(e);
System.exit(0);
}
}
sm = queue.pop();
}
if(stop){
return;
}
if (sm == null || !sm.isFull()) {
log.info("ShortMessage is null or is not valid");
continue;
}
log.info("will sender ShortMessage:"+sm);
sendShortMessage(sm);
}
}
队列结构:
public class SmsQueue {
private LinkedList<ShortMessage>[] queuePool;
@SuppressWarnings("unchecked")
public SmsQueue(int poolSize){
if(poolSize<=0||poolSize>20)
throw new ServiceException("初始化队列出错 应该 在0-20 而实际是"+poolSize);
queuePool = new LinkedList[poolSize];
for(int i=0;i<poolSize;i++){
queuePool[i] = new LinkedList<ShortMessage>();
}
}
/**
* 把短信放到队列尾
* @param shortMessage 短信 {@link #addFirst(ShortMessage)}
*/
public void push(ShortMessage shortMessage){
if(shortMessage!=null&&shortMessage.isFull()){
int index = shortMessage.getPriority();
synchronized (queuePool) {
if (index <= 0 || index >= queuePool.length) {
queuePool[0].addLast(shortMessage);
} else {
queuePool[index].addLast(shortMessage);
}
}
}
}
/**
* 从队列头拿出短信,如果整个队列为空,返回null
* @return
*/
public ShortMessage pop(){
synchronized (queuePool) {
for (int i = queuePool.length; i > 0; i--) {
if (!queuePool[i - 1].isEmpty()) {
return queuePool[i - 1].removeFirst();
}
}
}
return null;
}
/**
* 获得队列共有多少优先级
* @return
*/
public int getQueueWidth(){
synchronized(queuePool){
return queuePool.length;
}
}
/**
* 获得当前队列状态 没有返回空 list
* @return
*/
public List<QueueStatus> getStatus(){
List<QueueStatus> retList = new ArrayList<QueueStatus>();
int i=0;
synchronized (queuePool) {
for (LinkedList<ShortMessage> queue : queuePool) {
QueueStatus status = new QueueStatus();
status.setPrority(++i);
status.setSize(queue.size());
retList.add(status);
}
}
return retList;
}
/**
* 获得所有没有发送出去的对象,发回的list 中对象引用不能修改
* 由于对象已经在内存中 所以不考虑分页
* @return
*/
public List<ShortMessage> getAllShortMessage(){
List<ShortMessage> retList = new ArrayList<ShortMessage>();
synchronized (queuePool) {
for (LinkedList<ShortMessage> queue : queuePool) {
for (ShortMessage shortMessage : queue) {
retList.add(shortMessage);
}
}
}
return Collections.unmodifiableList(retList);
}
/**
* @return id ==null 或者没有找到 返回 false
* @param id
* @return
*/
public boolean removeSMById(String id){
if(id==null)
return false;
synchronized (queuePool) {
for (LinkedList<ShortMessage> queue : queuePool) {
for (ShortMessage shortMessage : queue) {
if(shortMessage.getId().equals(id)){
queue.remove(shortMessage);
return true;
}
}
}
return false;
}
}
/**
* 对列中共有多少短信
* @return
*/
public int totalShortMessageNum(){
int size=0;
synchronized(queuePool){
for(LinkedList list:queuePool){
if(!list.isEmpty()){
size+=list.size();
}
}
return size;
}
}
/**
* 队列是否为空
* @return
*/
public boolean isEmpty() {
synchronized (queuePool) {
for (LinkedList list : queuePool) {
if (!list.isEmpty()) {
return false;
}
}
}
return true;
}
/**
* 清空队列
*
*/
public void clear(){
synchronized (queuePool) {
for (LinkedList list : queuePool) {
list.clear();
}
}
}
/**
* 把短信放到队列头 (发送短信失败的时候使用)
* @param sm
*/
public void addFirst(ShortMessage sm) {
synchronized (queuePool) {
queuePool[queuePool.length - 1].addFirst(sm);
}
}
}
问题补充:
你在造轮子啊。。。。。
为什么说我在造轮子呢,有现成的东西可以使用么
========================
按照你的描述 问题在生产者和消费者的可能不太 看看发送模块是否有异样日志
正式环境下面消费者(发送短信)消费一次需要几秒甚至一分多
而我在测试环境下面消费者使用了mock
发送了几万条一点问题都没有
不知道什么原因
另外生产环境下消费者是其他厂商提供的api
该问题已经关闭: 超过15天由系统自动关闭
回答
按照你的描述 问题在生产者和消费者的可能不太 看看发送模块是否有异样日志
guoxu1231 (初级程序员) 2008-05-23
你在造轮子啊。。。。。
dongsj8325@sohu.com (初级程序员) 2008-05-23
已解决问题数: 523
待解决问题数: 280
已关闭问题数: 1029
待解决问题数: 280
已关闭问题数: 1029
问答分类
答题高手
- lggege 高级程序员 ( 48 - 587 )
- ham 资深程序员 ( 46 - 655 )
- congjl2002 资深程序员 ( 43 - 617 )
- llade 高级程序员 ( 32 - 554 )
- hjgundam 高级程序员 ( 30 - 440 )
- wangxin0072000 高级程序员 ( 27 - 396 )
- jasongreen 高级程序员 ( 26 - 528 )
- 温柔一刀 高级程序员 ( 23 - 336 )
- kyo100900 中级程序员 ( 11 - 209 )
- mewleo 中级程序员 ( 11 - 242 )




