java同步代码(synchronized)中使用BlockingQueue
2024-09-29 15:52:02
说起BlockingQueue，大家最熟悉的就是生产者-消费者模式下的应用。但是如果在调用queue的上层代码加了同步块就会导致线程死锁。
例如：
static BlockingQueue<String> queue = new LinkedBlockingQueue();
/**
* 同步锁
*/
static Object lock = new Object();
static void producer(){
synchronized (lock){
queue.put("1");
}
}
static void cosumer(){
synchronized (lock){
//一旦阻塞，将挂起当前线程，lock锁永远等不到释放，生产者也就无法添加元素，take也就永远阻塞
String msg = queue.take();
}
}
但是同步块必须使用的情况下，怎样改进queue的使用呢？见下面示例：
package com.hdwang; import com.alibaba.fastjson.JSON; import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue; /**
* Created by hdwang on 2018/4/17.
*/
public class MultiQueueSynTest { static BlockingQueue<Packet> queue1 = new LinkedBlockingQueue(); static BlockingQueue<Packet> queue2 = new LinkedBlockingQueue(); static int seq = 1; /**
* 同步锁
*/
static Object lock = new Object(); static void commit(String msg){
synchronized (lock) {
Packet packet = new Packet();
packet.setSeq(seq++);
packet.setMsg(msg);
try { //queue1.put(packet); //阻塞式添加元素 while(queue1.size()== Integer.MAX_VALUE){ //队满，等待
lock.wait();
} queue1.offer(packet); //非阻塞式添加元素即可
System.out.println("commit msg:" + JSON.toJSONString(packet));
lock.notifyAll(); //通知等待线程 } catch (InterruptedException e) {
e.printStackTrace();
}
}
} static void send(){
while(true) {
synchronized (lock) {
try { //Packet packet = queue1.take(); //阻塞式取元素
//queue2.put(packet); while(queue1.isEmpty()) { //队空，等待
lock.wait(); //等待，交出锁
} Packet packet = queue1.poll(); //非阻塞式取元素即可
System.out.println("send msg:" + JSON.toJSONString(packet));
lock.notifyAll(); //通知等待线程 while (queue2.size() == Integer.MAX_VALUE){ //队满，等待
lock.wait(); //等待，交出锁
}
queue2.offer(packet);
System.out.println("msg->queue2:"+JSON.toJSONString(packet));
lock.notifyAll(); //通知等待线程 } catch (InterruptedException e) {
e.printStackTrace();
}
}
}
} public static void main(String[] args) {
//生产者1
new Thread(new Runnable() {
@Override
public void run() {
while(true){ //不断产生消息
commit("hello1"); try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}).start();
//生产者2
new Thread(new Runnable() {
@Override
public void run() {
while(true){ //不断产生消息
commit("hello2"); try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}).start(); //消费者
new Thread(new Runnable() {
@Override
public void run() { send();
}
}).start(); } static class Packet{
int seq;
String msg; public int getSeq() {
return seq;
} public void setSeq(int seq) {
this.seq = seq;
} public String getMsg() {
return msg;
} public void setMsg(String msg) {
this.msg = msg;
}
} }
运行结果
commit msg:{"msg":"hello1","seq":1}
send msg:{"msg":"hello1","seq":1}
msg->queue2:{"msg":"hello1","seq":1}
commit msg:{"msg":"hello2","seq":2}
send msg:{"msg":"hello2","seq":2}
msg->queue2:{"msg":"hello2","seq":2}
commit msg:{"msg":"hello1","seq":3}
send msg:{"msg":"hello1","seq":3}
msg->queue2:{"msg":"hello1","seq":3}
commit msg:{"msg":"hello2","seq":4}
send msg:{"msg":"hello2","seq":4}
msg->queue2:{"msg":"hello2","seq":4}
commit msg:{"msg":"hello1","seq":5}
send msg:{"msg":"hello1","seq":5}
msg->queue2:{"msg":"hello1","seq":5}
commit msg:{"msg":"hello2","seq":6}
send msg:{"msg":"hello2","seq":6}
msg->queue2:{"msg":"hello2","seq":6}
最新文章
- android 闪屏还是会出现黑屏问题
- jar包目录下MANIFEST.MF标准格式
- php ob_start()、ob_end_flush和ob_end_clean()多级缓冲
- WebViewJavascriptBridge 原理分析
- linux mysql 授权以及 iptables开启3306
- NETSH WINSOCK RESET这条命令的含义和作用？
- (转)iOS7界面设计规范(4) - UI基础 - 布局
- Storm流计算从入门到精通之技术篇(高并发策略、批处理事务、Trident精解、运维监控、企业场景)
- Hibernate 系列教程12-继承-Join策略
- python入门小记
- JSON(五）——同步请求中使用JSON格式字符串进行交互（不太常见的用法）
- HttpClient4.5.2调用示例（转载+原创）
- 关于接口（Interface）
- Ubuntu Server 18.04 修改网路配置
- sed，grep，进阶+source+export+环境变量
- android周期性任务
- 方法调用时候 传入this 谁调用 传入谁
- 201621123012《Java程序设计》第12次学习总结
- BluetoothSocket详解
- Buildroot自动化交叉编译工具：其一【转】