线程同步模型, 生产者/消费者, 读写同步,线程池,concurrent map.

LZ的线程文章写得都不错，对理解线程还是很好的，支持

c++的pv问题偶写过,用Java的synchronized来搞还没尝试过

您好
我写的生产消费出现了问题,您看看是出在哪里了，谢谢


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

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

多线程通常是想利用多CPU的并发执行，如果线程执行粒度太小，不但不会提升反而降低性能，但粒度合适的话对性能肯定有很好的提升的