java中的消息队列怎么利用多线程实现?相信很多没有经验的人对此束手无策,为此本文总结了问题出现的原因和解决方法,通过这篇文章希望你能解决这个问题。
1、定义一个队列缓存池:
//static修饰的成员变量和成员方法独立于该类的任何对象。也就是说,它不依赖类特定的实例,被类的所有实例共享。
private static List<Queue> queueCache = new LinkedList<Queue>();
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<offerMaxQueue&&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中的消息队列怎么利用多线程实现的方法了吗?如果还想学到更多技能或想了解更多相关内容,欢迎关注亿速云行业资讯频道,感谢各位的阅读!
亿速云「云服务器」,即开即用、新一代英特尔至强铂金CPU、三副本存储NVMe SSD云盘,价格低至29元/月。点击查看>>
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。