生产者生产了短信，2个小时后消费者才将短信发送出去，问题出在了哪里

0 0

生产者生产了短信，2个小时后消费者才将短信发送出去，问题出在了哪里20

才发现这里有个问答模块

转发一下

整个算法大致是：
多个线程同时生产短信，放到队列中（存在优先级，且先进先出），
一个消费线程从队列中拿出短信发送出去

目前出现了一个现象
生产者生产了短信，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

企业应用