扫二维码与项目经理沟通
我们在微信上24小时期待你的声音
解答本文疑问/技术咨询/运营咨询/技术建议/互联网交流
java中的消息队列怎么利用多线程实现?相信很多没有经验的人对此束手无策,为此本文总结了问题出现的原因和解决方法,通过这篇文章希望你能解决这个问题。
网站建设哪家好,找成都创新互联!专注于网页设计、网站建设、微信开发、小程序开发、集团企业网站建设等服务项目。为回馈新老客户创新互联还提供了南京免费建站欢迎大家使用!
1、定义一个队列缓存池:
//static修饰的成员变量和成员方法独立于该类的任何对象。也就是说,它不依赖类特定的实例,被类的所有实例共享。 private static ListqueueCache = new LinkedList ();
2、定义队列缓冲池最大消息数,如果达到该值,那么队列检入将等待检出低于该值时继续进行。
private Integer offerMaxQueue = 2000;
3、定义检出线程,如果队列缓冲池没有消息,那么检出线程会线程等待中
new Thread(){ public void run(){ while(true){ String ip = null; try { synchronized (queueCache) { Integer size = queueCache.size(); if(size==0){ //队列缓存池没有消息,等待。。。。 queueCache.wait(); } Queue queue = queueCache.remove(0); if(isIpLock(queueStr)){//假若这个是一个多应用的分布式系统,那么这个判断应该是分布式锁,这里说的锁不是线程停止,而是跳过该消息,滞后处理 queueCache.add(queue);该queue重新加入队列缓冲池,滞后处理, continue; }else{ ;//这里是处理该消息的操作。 } size = queueCache.size(); if(size=0){ queueCache.notifyAll();//在队列缓存池不超过最大值的前提下,假若检入正在等待中,那么那么让他们排队检入。 } } } catch (Exception e) { e.printStackTrace(); }finally{ try {//检出该消息队列的锁 unIpLock(queueStr); } catch (Execption e) {//捕获异常,不能让线程挂掉 e.printStackTrace(); } } } }.start();
4、检入队列
synchronized (queueCache) { while(true){ Integer size = queueCache.size(); if(size>=offerMaxQueue){ try { queueCache.wait(); continue;//继续执行等待中的检入任务。 } catch (InterruptedException e) { e.printStackTrace(); } }//IF if(size<=offerMaxQueue&&size>0){ queueCache.notifyAll(); } break;//检入完毕 }//while }
5、锁方法实现
/** * 锁 * @param ip * @return * @throws */ public Boolean isLock(String queueStr) { return this.redisManager.setnx(queueStr+"_lock", "LOCK", 10000)!=1; } //解锁 public void unIpLock(String queueStr) { if(ip!=null){ this.redisManager.del(queueStr+"_lock"); // lock.unlock(); } }
看完上述内容,你们掌握java中的消息队列怎么利用多线程实现的方法了吗?如果还想学到更多技能或想了解更多相关内容,欢迎关注创新互联行业资讯频道,感谢各位的阅读!
我们在微信上24小时期待你的声音
解答本文疑问/技术咨询/运营咨询/技术建议/互联网交流