目录
一、引言
二、生产者
三、消费者
四、扩展
五、总结
一、引言
本篇文章就是本次Java项目的最后一篇文章了，本篇文章主要介绍基于MQ的生产者消费者模型的代码编写
二、生产者
public class DemoConsumer {public static void main(String[] args) throws IOException, InterruptedException, MqException {System.out.println("启动消费者！");ConnectionFactory factory = new ConnectionFactory();factory.setHost("127.0.0.1");factory.setPort(9090);Connection connection = factory.newConnection();Channel channel = connection.createChannel();channel.exchangeDeclare("testExchange", ExchangeType.DIRECT,true,false,null);channel.queueDeclare("testQueue",true,false,false,null);channel.basicConsume("testQueue", true, new Consumer() {@Overridepublic void handleDelivery(String consumeTag, BasicProperties basicProperties, byte[] body) throws MqException, IOException {System.out.println("[消费数据]开始！");System.out.println("consumerTag="+consumeTag);System.out.println("basicProperties="+basicProperties);String bodyString = new String(body,0, body.length);System.out.println("body="+bodyString);System.out.println("[消费数据]结束！");}});while (true){Thread.sleep(500);}}
}三、消费者
public class DemoProducer {public static void main(String[] args) throws IOException, InterruptedException {System.out.println("启动生产者！");ConnectionFactory factory = new ConnectionFactory();factory.setHost("127.0.0.1");factory.setPort(9090);Connection connection = factory.newConnection();Channel channel = connection.createChannel();channel.exchangeDeclare("testExchange", ExchangeType.DIRECT,true,false,null);channel.queueDeclare("testQueue",true,false,false,null);byte[] body = "hello".getBytes();boolean ok = channel.basicPublish("testExchange","testQueue",null,body);System.out.println("消息投递完成！ok="+ok);Thread.sleep(500);channel.close();connection.close();}
}四、扩展
五、总结
本次Java项目“模拟消息队列”就全部完成了，感兴趣的小伙伴可以主机敲一遍代码以及实现上述的拓展API，全部代码我已上传到gitee：spring-mq · 别像我这么帅/Javacode - 码云 - 开源中国，感谢观看！