长沙网站开发智,做cpa网站,中国建设工程信息网官网查询系统,重庆网站关键词排名一丶什么是MQ Message Queue(消息队列#xff09;简称MQ#xff0c;是一种应用程序对应用程序的消息通信机制。在MQ中#xff0c;消息以队列形式存储#xff0c;以便于异步传输#xff0c;在MQ中#xff0c;发布者#xff08;生产者#xff09;将消息放入队列#xff…一丶什么是MQ Message Queue(消息队列简称MQ是一种应用程序对应用程序的消息通信机制。在MQ中消息以队列形式存储以便于异步传输在MQ中发布者生产者将消息放入队列而消费者从队列中读取并处理这些消息这种设计允许生产者和消费者之间解耦提高系统的响应速度和吞吐量MQ常用于解耦系统之间的依赖关系提高系统的稳定性和可扩展性MQ还支持消峰即以稳定的系统资源应对突发的流量冲剂然而使用MQ也可能带来一些挑战如系统可用性降低、系统复杂度提高、以及消息一致性问题等
二丶常见MQ有哪些 目前业界有很多的MQ产品例如RabbitMQ、RocketMQ、Kafka、ZeroMQ、MetaMQ等也有直接使用Redis充当消息队列的案列而这些消息队列产品各有侧重在实际选型时需要结合自身需求及MQ的产品特征等综合考虑 RabbitMQ是实现了高级消息队列协议(AMQP)的开源消息代理软件亦称面向消息的中间件。RabbitMQ服务器是Erlang语言编写而集群和故障转移是构建在开放电信平台框架上的。所有的主要变成语言均有与代理接口通讯的客户端库。
三丶RabbitMQ六种消息模式 1.Simple Work Queue简单工作队列常见的一对一模式一条消息由一个消费者进行消费。如果有多个消费者默认是使用轮询的方式将消息分配消费者 2.Work Queues工作队列模式也叫公屏队列能者多劳的消息队列模型。队列必须收到来自消费者的手动ACK消息确认机制才可以继续往消费者发送消息。 3.Publish/Subscribe发布订阅模式一条消息被多和消费者消费。 4.Ruoting路由模式有选择的接收消息。 5.Topics主题模式通过一定的规则来选择性的接收消息 6.RPC模式发布者发布消息并且通过RPC方式等待结果。 1Simple Work Queue简单工作队列 消息生产后将消息放入队列消息的消费者consumer监听消息队列嘛如果队列中有消息就消费掉消息被消费后自动从消息队列中删除。也可能出现异常
/*** Description:获取RabbitMQ连接* Author: xy丶*/
public class RabbitMQConnection {public final static String RABBITMQ_SERVER_HOST 192.168.0.122;public final static int RABBITMQ_SERVER_PORT 5672;public final static String VIRTUAL_HOST /XY_HOST;public static Connection getConnection() throws IOException, TimeoutException {//1、创建连接ConnectionFactory factory new ConnectionFactory();//2、设置主机名factory.setHost(RABBITMQ_SERVER_HOST);//3、设置通讯端口默认是5672不专门设置也可以factory.setPort(RABBITMQ_SERVER_PORT);//4、设置账号和密码factory.setUsername(admin);factory.setPassword(admin);//4、设置Virtual Hostfactory.setVirtualHost(VIRTUAL_HOST);//5、创建连接return factory.newConnection();}} 消费者
/*** Description:简单工作队列模式消费者* Author: xy丶*/
Slf4j
public class Consumer {private final static String SIMPLE_WORK_QUEUE simple_work_queue;public static void main(String[] args) throws IOException, InterruptedException, TimeoutException {//获取连接Connection connection RabbitMQConnection.getConnection();//获取通道Channel channel connection.createChannel();channel.queueDeclare(SIMPLE_WORK_QUEUE, false, false, false, null);QueueingConsumer consumer new QueueingConsumer(channel);channel.basicConsume(SIMPLE_WORK_QUEUE, true, consumer);QueueingConsumer.Delivery delivery consumer.nextDelivery();String message new String(delivery.getBody());log.info(Consumer reception message);}
}
生产者
/*** Description:简单工作队列模式生产者* Author: xy丶*/
Slf4j
public class Producer {private final static String SIMPLE_WORK_QUEUE simple_work_queue;/*** 简单工作队列模式* param args* throws IOException* throws TimeoutException*/public static void main(String[] args) throws IOException, TimeoutException {//创建连接Connection connection RabbitMQConnection.getConnection();//创建通道Channel channel connection.createChannel();// 声明队列 String var1, 是否持久化// boolean var2, 是否排外 只允许该channel访问该队列 为true只能有一个消费者// boolean var3, 是否自动删除// boolean var4, 消费完删除// MapString, Object var5 其他属性channel.queueDeclare(SIMPLE_WORK_QUEUE, false, false, false, null);// 消息内容 String var1, 是否持久化// boolean var2, 是否排外 只允许该channel访问该队列 为true只能有一个消费者// boolean var3, 是否自动删除// boolean var4, 消费完删除// MapString, Object var5 其他属性String message Hello Word!!!;channel.basicPublish(, SIMPLE_WORK_QUEUE,null,message.getBytes());log.info(Producer send message);//最后关闭通关和连接channel.close();connection.close();} (2).Work Queues工作队列模式 创建生产者两个消费者看看效果 生产者
package com.puwang.MQ.workQueue;import com.puwang.MQ.config.RabbitMQConnection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import lombok.extern.slf4j.Slf4j;import java.io.IOException;
import java.util.Scanner;
import java.util.concurrent.TimeoutException;/*** Description:工作队列模式生产者* Author: xy丶*/
Slf4j
public class Producer {private final static String QUEUE_WORK QUEUE_WORK;public static void main(String[] args) throws Exception {Connection connection RabbitMQConnection.getConnection();Channel channel connection.createChannel();channel.queueDeclare(QUEUE_WORK, false, false, false, null);for(int i 0; i 20; i){String message 娃哈哈 i;channel.basicPublish(, QUEUE_WORK, null, message.getBytes());System.out.println(sendmessage);Thread.sleep(i*10);}channel.close();connection.close();}
}
Slf4j
public class WorkQueueConsumer1 {private final static String QUEUE_WORK QUEUE_WORK;/*** 结果** 1、一条消息只会被一个消费者接收** 2、rabbit采用轮询的方式将消息是平均发送给消费者的** 3、消费者在处理完某条消息后才会收到下一条消息。* param args* throws IOException* throws TimeoutException*/public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {Connection connection RabbitMQConnection.getConnection();Channel channel connection.createChannel();channel.queueDeclare(QUEUE_WORK, false,false, false,null);//同一时刻服务器只会发送一条消息给消费者channel.basicQos(1);QueueingConsumer consumer new QueueingConsumer(channel);//关于手工确认 待之后有时间研究下channel.basicConsume(QUEUE_WORK, false, consumer);while(true){QueueingConsumer.Delivery delivery consumer.nextDelivery();String message new String(delivery.getBody());log.info([消费者1] Received1 message);Thread.sleep(10);//返回确认状态channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);}}
}Slf4j
public class WorkQueueConsumer2 {private final static String QUEUE_WORK QUEUE_WORK;public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {Connection connection RabbitMQConnection.getConnection();Channel channel connection.createChannel();channel.queueDeclare(QUEUE_WORK, false,false, false,null);//同一时刻服务器只会发送一条消息给消费者channel.basicQos(1);QueueingConsumer consumer new QueueingConsumer(channel);//关于手工确认 待之后有时间研究下channel.basicConsume(QUEUE_WORK, false, consumer);while(true){QueueingConsumer.Delivery delivery consumer.nextDelivery();String message new String(delivery.getBody());log.info([消费者2] Received1 message);Thread.sleep(10);//返回确认状态channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);}}
} (3).Publish/Subscribe发布订阅模式 生产者
/*** 订阅模式 生产者* 订阅模式一个生产者发送的消息会被多个消费者获取。* 消息产生者将消息放入交换机,交换机发布订阅把消息发送到所有消息队列中,对应消息队列的消费者拿到消息进行消费* 相关场景:邮件群发,群聊天,广播(广告)* Description:发布订阅模式生产者* Author: xy丶*/
Slf4j
public class Producer {private final static String PUBLISH_SUBSCRIBE_EXCHANGE Publish_subscribe_exchange;/*** 交换机发布订阅把消息发送到所有消息队列中,对应消息队列的消费者拿到消息进行消费* 相关场景:邮件群发,群聊天,广播(广告)* param args*/public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {//获取连接Connection connection RabbitMQConnection.getConnection();//从连接中获取一个通道Channel channel connection.createChannel();消费者绑定交换机 参数1 队列 参数2 交换机 参数3 routingKeychannel.exchangeDeclare(PUBLISH_SUBSCRIBE_EXCHANGE, fanout);//发送消息for (int i 0; i 10; i) {String message 哇哈哈哈i;log.info(send message message);//发送消息channel.basicPublish(PUBLISH_SUBSCRIBE_EXCHANGE, , null, message.getBytes(utf-8));Thread.sleep(100 * i);}channel.close();connection.close();}
}
消费者
/*** Description:发布订阅模式消费者* Author: xy丶*/
Slf4j
public class PublishSubscribeConsumer1 {//交换机名称private final static String PUBLISH_SUBSCRIBE_EXCHANGE Publish_subscribe_exchange;//队列名称private static final String PUBLISH_SUBSCRIBE_QUEUE publish_subscribe_queue;public static void main(String[] args) throws IOException, TimeoutException {//获取连接Connection connection RabbitMQConnection.getConnection();//从连接中获取一个通道final Channel channel connection.createChannel();//声明交换机分发:发布/订阅模式channel.exchangeDeclare(PUBLISH_SUBSCRIBE_EXCHANGE, fanout);//声明队列channel.queueDeclare(PUBLISH_SUBSCRIBE_QUEUE, false, false, false, null);//将队列绑定到交换机channel.queueBind(PUBLISH_SUBSCRIBE_QUEUE, PUBLISH_SUBSCRIBE_EXCHANGE, );//保证一次只分发一个int prefetchCount 1;channel.basicQos(prefetchCount);//定义消费者DefaultConsumer consumer new DefaultConsumer(channel) {//当消息到达时执行回调方法Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String message new String(body, utf-8);log.info([PublishSubscribeConsumer1] Receive message message);try {//消费者休息2s处理业务Thread.sleep(1000);}catch (InterruptedException e) {e.printStackTrace();}finally {System.out.println([1] done);//手动应答channel.basicAck(envelope.getDeliveryTag(), false);}}};//设置手动应答boolean autoAck false;//监听队列channel.basicConsume(PUBLISH_SUBSCRIBE_QUEUE, autoAck, consumer);}
}/*** Description:发布订阅模式消费者* Author: xy丶*/
Slf4j
public class PublishSubscribeConsumer1 {//交换机名称private final static String PUBLISH_SUBSCRIBE_EXCHANGE Publish_subscribe_exchange;//队列名称private static final String PUBLISH_SUBSCRIBE_QUEUE publish_subscribe_queue;public static void main(String[] args) throws IOException, TimeoutException {//获取连接Connection connection RabbitMQConnection.getConnection();//从连接中获取一个通道final Channel channel connection.createChannel();//声明交换机分发:发布/订阅模式channel.exchangeDeclare(PUBLISH_SUBSCRIBE_EXCHANGE, fanout);//声明队列channel.queueDeclare(PUBLISH_SUBSCRIBE_QUEUE, false, false, false, null);//将队列绑定到交换机channel.queueBind(PUBLISH_SUBSCRIBE_QUEUE, PUBLISH_SUBSCRIBE_EXCHANGE, );//保证一次只分发一个int prefetchCount 1;channel.basicQos(prefetchCount);//定义消费者DefaultConsumer consumer new DefaultConsumer(channel) {//当消息到达时执行回调方法Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String message new String(body, utf-8);log.info([PublishSubscribeConsumer1] Receive message message);try {//消费者休息2s处理业务Thread.sleep(1000);}catch (InterruptedException e) {e.printStackTrace();}finally {System.out.println([1] done);//手动应答channel.basicAck(envelope.getDeliveryTag(), false);}}};//设置手动应答boolean autoAck false;//监听队列channel.basicConsume(PUBLISH_SUBSCRIBE_QUEUE, autoAck, consumer);}
}/*** Description:发布订阅模式消费者* Author: xy丶*/
Slf4j
public class PublishSubscribeConsumer2 {//交换机名称private final static String PUBLISH_SUBSCRIBE_EXCHANGE Publish_subscribe_exchange;//队列名称private static final String PUBLISH_SUBSCRIBE_QUEUE publish_subscribe_queue;public static void main(String[] args) throws IOException, TimeoutException {//获取连接Connection connection RabbitMQConnection.getConnection();//从连接中获取一个通道final Channel channel connection.createChannel();//声明交换机分发:发布/订阅模式channel.exchangeDeclare(PUBLISH_SUBSCRIBE_EXCHANGE, fanout);//声明队列channel.queueDeclare(PUBLISH_SUBSCRIBE_QUEUE, false, false, false, null);//将队列绑定到交换机channel.queueBind(PUBLISH_SUBSCRIBE_QUEUE, PUBLISH_SUBSCRIBE_EXCHANGE, );//保证一次只分发一个int prefetchCount 1;channel.basicQos(prefetchCount);//定义消费者DefaultConsumer consumer new DefaultConsumer(channel) {//当消息到达时执行回调方法Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String message new String(body, utf-8);log.info([PublishSubscribeConsumer2] Receive message message);try {//消费者休息2s处理业务Thread.sleep(1000);}catch (InterruptedException e) {e.printStackTrace();}finally {System.out.println([1] done);//手动应答channel.basicAck(envelope.getDeliveryTag(), false);}}};//设置手动应答boolean autoAck false;//监听队列channel.basicConsume(PUBLISH_SUBSCRIBE_QUEUE, autoAck, consumer);}
} (4).Ruoting路由模式
消费者
/*** 1消息生产者将消息发送给交换机按照路由判断,路由是字符串(info) 当前产生的消息携带路由字符(对象的方法),交换机根据路由的key,只能匹配上路由key对应的消息队列,对应的消费者才能消费消息* 2根据业务功能定义路由字符串* 3从系统的代码逻辑中获取对应的功能字符串,将消息任务扔到对应的队列中业务场景:error 通知;EXCEPTION;错误通知的功能;传统意义的错误通知;* 客户通知;利用key路由,可以将程序中的错误封装成消息传入到消息队列中,开发者可以自定义消费者,实时接收错误* Description:发布订阅模式生产者* Author: xy丶*/
Slf4j
public class Producer {//路由交换机名称static final String ROUTING_DIRECT_EXCHANGE Routing_direct_exchange;//队列名称1 发送static final String QUEUE_SEND queue_send;//队列名称2 接收static final String QUEUE_RECEIVE queue_receive;public static void main(String[] args) throws Exception {//创建连接Connection connection RabbitMQConnection.getConnection();//创建频道Channel channel connection.createChannel();/*** 声明交换机* 参数1交换机名称* 参数2交换机类型fanout,toppic,direct,headers*/channel.exchangeDeclare(ROUTING_DIRECT_EXCHANGE, direct);/*** 声明创建队列* 参数1队列名称* 参数2是否定义持久化队列* 参数3是否独占本次连接* 参数4是否在不使用的时候自动删除队列* 参数5队列其它参数*/channel.queueDeclare(QUEUE_SEND,true,false,false,null);channel.queueDeclare(QUEUE_RECEIVE,true,false,false,null);/*** 队列绑定交换机* 参数1队列名称* 参数2是否定义持久化队列* 参数3是否独占本次连接*/channel.queueBind(QUEUE_SEND,ROUTING_DIRECT_EXCHANGE,send);channel.queueBind(QUEUE_RECEIVE,ROUTING_DIRECT_EXCHANGE,receive);//发送消息String message 路由模式routing key 为 send;/*** 参数1交换机名称如果没有指定则使用默认Default Exchage* 参数2路由key,简单模式可以传递队列名称* 参数3消息其它属性* 参数4消息内容*/channel.basicPublish(ROUTING_DIRECT_EXCHANGE,send,null,message.getBytes());log.info(已发送消息message);//发送消息message 路由模式routing key 为 receive;/*** 参数1交换机名称如果没有指定则使用默认Default Exchage* 参数2路由key,简单模式可以传递队列名称* 参数3消息其它属性* 参数4消息内容*/channel.basicPublish(ROUTING_DIRECT_EXCHANGE,receive,null,message.getBytes());log.info(已发送消息message);//关闭资源channel.close();connection.close();}
}
消费者 /***路由消费者*/
Slf4j
public class RoutingConsumer1 {//路由交换机名称static final String ROUTING_DIRECT_EXCHANGE Routing_direct_exchange;//队列名称1 发送static final String QUEUE_SEND queue_send;public static void main(String[] args) throws Exception {//创建连接Connection connection RabbitMQConnection.getConnection();//创建通道频道Channel channel connection.createChannel();//声明交换机channel.exchangeDeclare(ROUTING_DIRECT_EXCHANGE,direct);/*** 声明创建队列* 参数1队列名称* 参数2是否定义持久化队列* 参数3是否独占本次连接* 参数4是否在不使用的时候自动删除队列* 参数5队列其它参数*/channel.queueDeclare(QUEUE_SEND,true,false,false,null);//队列绑定交换机channel.queueBind(QUEUE_SEND, ROUTING_DIRECT_EXCHANGE,send);//创建消费这并设置消息处理DefaultConsumer consumer new DefaultConsumer(channel) {/*** consumerTag 消息者标签在channel.basicConsume时候可以指定* envelope 消息包的内容可从中获取消息id消息routingkey交换机* 消息和重传标志(收到消息失败后是否需要重新发送)* properties 属性信息* body 消息*/Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {//路由keylog.info(路由key为 envelope.getRoutingKey());//交换机log.info(交换机为 envelope.getExchange());//消息idlog.info(消息id为 envelope.getDeliveryTag());//收到的消息log.info(消费者1-接收到的消息为 new String(body, utf8));}};/*** 监听消息* 参数1队列名称* 参数2是否自动确认设置为true为表示消息接收到自动向mq回复接收到了mq接收到回复会删除消息设置为false则需要手动确认* 参数3消息接收到后回调*/channel.basicConsume(QUEUE_SEND, true, consumer);}/***路由消费者*/
Slf4j
public class RoutingConsumer2 {//路由交换机名称static final String ROUTING_DIRECT_EXCHANGE Routing_direct_exchange;//队列名称1 发送static final String QUEUE_RECEIVE queue_receive;public static void main(String[] args) throws Exception {//创建连接Connection connection RabbitMQConnection.getConnection();//创建通道频道Channel channel connection.createChannel();//声明交换机channel.exchangeDeclare(ROUTING_DIRECT_EXCHANGE,direct);/*** 声明创建队列* 参数1队列名称* 参数2是否定义持久化队列* 参数3是否独占本次连接* 参数4是否在不使用的时候自动删除队列* 参数5队列其它参数*/channel.queueDeclare(QUEUE_RECEIVE,true,false,false,null);//队列绑定交换机channel.queueBind(QUEUE_RECEIVE, ROUTING_DIRECT_EXCHANGE,receive);//创建消费这并设置消息处理DefaultConsumer consumer new DefaultConsumer(channel) {/*** consumerTag 消息者标签在channel.basicConsume时候可以指定* envelope 消息包的内容可从中获取消息id消息routingkey交换机* 消息和重传标志(收到消息失败后是否需要重新发送)* properties 属性信息* body 消息*/Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {//路由keylog.info(路由key为 envelope.getRoutingKey());//交换机log.info(交换机为 envelope.getExchange());//消息idlog.info(消息id为 envelope.getDeliveryTag());//收到的消息log.info(消费者2-接收到的消息为 new String(body, utf8));}};/*** 监听消息* 参数1队列名称* 参数2是否自动确认设置为true为表示消息接收到自动向mq回复接收到了mq接收到回复会删除消息设置为false则需要手动确认* 参数3消息接收到后回调*/channel.basicConsume(QUEUE_RECEIVE, true, consumer);}
} (5).Topics主题模式/路由模式的一种
生产者
/*** 跟 routing 路由模式类似只不过路由模式是指定固定的路由键 routingKey而主题模式是可以模糊匹配路由routingKey类似于SQL中 和 like 的关系* 消息可能匹配多个消费者但是同一个队列的中的消息不会被重复消费**要求* Topic 模式消息的 routing_key 不能随意写必须满足一定的要求它必须是一个单词列表以 “.” 或者 “#” 分隔开。这些单词可以是任意单词这个单词列表最多不能超过 255 个字节。* 分隔符* “*(星号)”可以代替一个单词* “#(井号)”可以替代零个或多个单词* Description:主题模式* Author: xy丶*/
Slf4j
public class TopicProducer {public static final String TOPIC_EXCHANGE topic_exchange;public static final String TOPIC_QUEUE_ONE topic_queue_one;public static final String TOPIC_QUEUE_TWO topic_queue_two;public static void main(String[] args) throws Exception {//声明用作全局变量的队列变量和交换价变量//创建连接Connection connection RabbitMQConnection.getConnection();//创建信道Channel channel connection.createChannel();//声明队列channel.queueDeclare(TOPIC_QUEUE_ONE,true,false,false,null);channel.queueDeclare(TOPIC_QUEUE_TWO,true,false,false,null);//声明交换机channel.exchangeDeclare(TOPIC_EXCHANGE, topic,true);//绑定队列channel.queueBind(TOPIC_QUEUE_ONE,TOPIC_EXCHANGE,*.orange.*);channel.queueBind(TOPIC_QUEUE_TWO,TOPIC_EXCHANGE,*.*.rabbit);channel.queueBind(TOPIC_QUEUE_TWO,TOPIC_EXCHANGE,lazy.#);//发生消息for (int i 0; i 10 ; i) {String msggoodnight!My love worldi;channel.basicPublish(TOPIC_EXCHANGE,ag.we.rabbit,null,msg.getBytes());}}
}
消费者
Slf4j
public class TopicCustomer1 {public static final String TOPIC_QUEUE_ONEtopic_queue_one;public static void main(String[] args) throws Exception{//创建连接Connection connection RabbitMQConnection.getConnection();//创建信道Channel channel connection.createChannel();DefaultConsumer consumer new DefaultConsumer(channel){Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {log.info(TopicCustomer1:new String(body));}};channel.basicConsume(TOPIC_QUEUE_ONE,true,consumer);}
}Slf4j
public class TopicCustomer2 {public static final String TOPIC_QUEUE_TWOtopic_queue_two;public static void main(String[] args) throws Exception{//创建连接Connection connection RabbitMQConnection.getConnection();//创建信道Channel channel connection.createChannel();DefaultConsumer consumer new DefaultConsumer(channel){Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {log.info(TopicCustomer22:new String(body));}};channel.basicConsume(TOPIC_QUEUE_TWO,true,consumer);}
}Slf4j
public class TopicCustomer3 {public static final String TOPIC_QUEUE_TWO topic_queue_two;public static void main(String[] args) throws Exception{//创建连接Connection connection RabbitMQConnection.getConnection();//创建信道Channel channel connection.createChannel();DefaultConsumer consumer new DefaultConsumer(channel){Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {log.info(TopicCustomer2:new String(body));}};channel.basicConsume(TOPIC_QUEUE_TWO,true,consumer);}
} (6).RPC模式
