java线程实现简单消息队列 1对1模式

2019-11-08 02:42:48

字体：大中小

来源：转载

供稿：网友

用java线程模拟生产消费者

创建线程Pone 类 set获取队列中的消息

package com.test.mq.one_one;

public class Pone {

PRivate String lock;public Pone(String lock){super();this.lock = lock;}public void setValue(){try {synchronized (lock) {if(!ValueObject.value.equals("")){//判断生产者是否存放消息进来，存放等待，反之存入消息lock.wait();}String value = System.currentTimeMillis()+"_"+System.nanoTime();System.out.println("set = "+value);ValueObject.value = value;lock.notify();}} catch (Exception e) {e.printStackTrace();}}

}

创建线程Cone 类 get获取队列中的消息

package com.test.mq.one_one;public class Cone {private String lock;public Cone(String lock){super();this.lock = lock;}public void getValue(){try {synchronized (lock) {//枷锁lock取得对象锁，P生产线程等待if(ValueObject.value.equals("")){//判断消费是否被存放进来，如果未存放一直等待，反之取出队列并清除队列消息lock.wait();}System.out.println("get = "+ValueObject.value);ValueObject.value = "";lock.notify();}} catch (Exception e) {// TODO: handle exception}}}

创建C线程消费者 get队列里面的消息

package com.test.mq.one_one.thread;import com.test.mq.one_one.Cone;public class C implements Runnable{private Cone cone;public C(Cone cone){this.cone = cone;}public void run() {while (true) {

try {//为了更好后台查看打印记录让线程停顿一秒Thread.sleep(1000);} catch (InterruptedException e) {// TODO Auto-generated catch blocke.printStackTrace();}cone.getValue();}}}

创建消费者P 生产者队列里面存放消息

package com.test.mq.one_one.thread;import com.test.mq.one_one.Pone;public class P implements Runnable {private Pone pone;public P(Pone pone){this.pone = pone;}@Overridepublic void run() {while (true) {

try {Thread.sleep(1000);} catch (InterruptedException e) {// TODO Auto-generated catch blocke.printStackTrace();}pone.setValue();}}}

设置消息队列存放类

package com.test.mq.one_one;public class ValueObject {protected static String value = "";}

main方法

package com.test.mq.one_one;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import com.test.mq.one_one.thread.C;import com.test.mq.one_one.thread.P;public class MqOneTest {public static void main(String[] args) {ExecutorService fixedThreadPool = Executors.newFixedThreadPool(2); String lock = System.in.toString();Pone p = new Pone(lock);Cone c = new Cone(lock);fixedThreadPool.execute(new P(p));fixedThreadPool.execute(new C(c));}}

打印输出