RabbitMQ的五大常用模式

RabbitMQ 的“五大工作模式”是基于其核心组件交换机、队列、绑定的不同组合方式而形成的。前两种模式简单、工作队列实际上使用的是 RabbitMQ 内置的默认交换机而后三种模式发布/订阅、路由、主题则需要显式地声明特定类型的交换机来实现灵活的消息分发一.简单模式结构一个生产者P将消息直接发送到队列一个消费者C从该队列获取消息。核心特点消息被消费后即刻从队列移除一条消息只会被一个消费者接收。步骤1.创建工程生产者消费者2.分别添加依赖3.编写生产者发送消息4.编写消费者接收消息生产者代码public class producer_HelloWorld { public static void main(String[] args) throws IOException, TimeoutException { //1.创建链接工厂 ConnectionFactory factory new ConnectionFactory(); //2.设置参数 factory.setHost(127.0.0.1);//ip地址 factory.setPort(5672); factory.setVirtualHost(/idcast);//虚拟机 factory.setUsername(yang); factory.setPassword(yang); //3.创建连接 Connection Connection connection factory.newConnection(); //4.创建channel Channel channel connection.createChannel(); //5.创建队列queue //队列名称是否持久化当mq重启后还在是否独占只能有一个消费者监听这个队列。是否自动删除参数 //如果没有一个名字是hello world的队列则会创建该队列如果有则不会创建 channel.queueDeclare(hello world,true,false,false,null); //6.发送消息 /** * exchange:交换机名称简单模式下交换机会使用默认的“” * routingkey:路由名称 * props:配置信息 * body:发送消息数据 */ String body hello rabbitmq; channel.basicPublish(,hello world,null,body.getBytes()); //7.释放资源 先关 channel再关 connection顺序反了会报错 channel.close(); connection.close(); } } 创建连接工厂----设置参数----创建连接Connection----创建channel----创建队列queue----发送消息----释放资源消费者代码public class comsume_HelloWorld { public static void main(String[] args) throws IOException, TimeoutException { //1.创建连接工厂 ConnectionFactory factory new ConnectionFactory(); //2.设置参数 factory.setHost(127.0.0.1);//ip地址 factory.setPort(5672); factory.setVirtualHost(/idcast);//虚拟机 factory.setUsername(yang); factory.setPassword(yang); //3.创建连接 Connection Connection connection factory.newConnection(); //4.创建channel Channel channel connection.createChannel(); //5.创建队列queue //队列名称是否持久化当mq重启后还在是否独占只能有一个消费者监听这个队列。是否自动删除参数 //如果没有一个名字是hello world的队列则会创建该队列如果有则不会创建 channel.queueDeclare(hello world,true,false,false,null); //6.接收消息 /** * queue:队列名称 * autoAck:是否自动确认 * callback:回调对象 */ Consumer comsumer new DefaultConsumer(channel){ Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { super.handleDelivery(consumerTag, envelope, properties, body); System.out.println(consumerTag:consumerTag); System.out.println(envelope:envelope.getExchange()); System.out.println(envelope:envelope.getRoutingKey()); System.out.println(properties:properties); System.out.println(body:new String(body)); } }; channel.basicConsume(hello world,true,comsumer); } } 创建连接工厂----设置参数----创建连接Connection----创建channel----创建队列queue----接收消息二.工作队列模式结构一个生产者对应一个队列但有多个消费者C1, C2...同时监听同一个队列。核心特点消息被竞争消费默认轮询分发即 C1 拿一条C2 拿一条。这意味着同一条消息只能被其中一个消费者获取主要用于负载均衡或处理耗时任务如抢票、发短信。生产者代码public class producer_WorkQueues { public static void main(String[] args) throws IOException, TimeoutException { //1.创建链接工厂 ConnectionFactory factory new ConnectionFactory(); //2.设置参数 factory.setHost(127.0.0.1);//ip地址 factory.setPort(5672); factory.setVirtualHost(/idcast);//虚拟机 factory.setUsername(yang); factory.setPassword(yang); //3.创建连接 Connection Connection connection factory.newConnection(); //4.创建channel Channel channel connection.createChannel(); //5.创建队列queue //队列名称是否持久化当mq重启后还在是否独占只能有一个消费者监听这个队列。是否自动删除参数 //如果没有一个名字是hello world的队列则会创建该队列如果有则不会创建 channel.queueDeclare(work_queues,true,false,false,null); //6.发送消息 /** * exchange:交换机名称简单模式下交换机会使用默认的“” * routingkey:路由名称 * props:配置信息 * body:发送消息数据 */ for (int i 0; i 10; i) { String body ihello rabbitmq; channel.basicPublish(,work_queues,null,body.getBytes()); } //7.释放资源 先关 channel再关 connection顺序反了会报错 channel.close(); connection.close(); } }消费者代码public class comsume_WorkQueues1 { public static void main(String[] args) throws IOException, TimeoutException { //1.创建链接工厂 ConnectionFactory factory new ConnectionFactory(); //2.设置参数 factory.setHost(127.0.0.1);//ip地址 factory.setPort(5672); factory.setVirtualHost(/idcast);//虚拟机 factory.setUsername(yang); factory.setPassword(yang); //3.创建连接 Connection Connection connection factory.newConnection(); //4.创建channel Channel channel connection.createChannel(); //5.创建队列queue //队列名称是否持久化当mq重启后还在是否独占只能有一个消费者监听这个队列。是否自动删除参数 //如果没有一个名字是hello world的队列则会创建该队列如果有则不会创建 channel.queueDeclare(work_queues,true,false,false,null); //6.接收消息 /** * queue:队列名称 * autoAck:是否自动确认 * callback:回调对象 */ Consumer comsumer new DefaultConsumer(channel){ Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { super.handleDelivery(consumerTag, envelope, properties, body); // System.out.println(consumerTag:consumerTag); // System.out.println(envelope:envelope.getExchange()); // System.out.println(envelope:envelope.getRoutingKey()); // System.out.println(properties:properties); System.out.println(body:new String(body)); } }; channel.basicConsume(work_queues,true,comsumer); } }public class comsume_WorkQueues2 { public static void main(String[] args) throws IOException, TimeoutException { //1.创建链接工厂 ConnectionFactory factory new ConnectionFactory(); //2.设置参数 factory.setHost(127.0.0.1);//ip地址 factory.setPort(5672); factory.setVirtualHost(/idcast);//虚拟机 factory.setUsername(yang); factory.setPassword(yang); //3.创建连接 Connection Connection connection factory.newConnection(); //4.创建channel Channel channel connection.createChannel(); //5.创建队列queue //队列名称是否持久化当mq重启后还在是否独占只能有一个消费者监听这个队列。是否自动删除参数 //如果没有一个名字是hello world的队列则会创建该队列如果有则不会创建 channel.queueDeclare(work_queues,true,false,false,null); //6.接收消息 /** * queue:队列名称 * autoAck:是否自动确认 * callback:回调对象 */ Consumer comsumer new DefaultConsumer(channel){ Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { super.handleDelivery(consumerTag, envelope, properties, body); // System.out.println(consumerTag:consumerTag); // System.out.println(envelope:envelope.getExchange()); // System.out.println(envelope:envelope.getRoutingKey()); // System.out.println(properties:properties); System.out.println(body:new String(body)); } }; channel.basicConsume(work_queues,true,comsumer); } }注是多个消费者三.Pub/Sub订阅发布模式结构生产者将消息发送到 Fanout 交换机。该交换机会将消息无视路由键Routing Key广播给所有与之绑定的队列。每个队列对应一个消费者。核心特点一条消息会被多个消费者同时接收。适用于“国家气象局发天气百度/腾讯/网易同时接收”这种一对多的广播场景。区别在生产者和消息队列之间加了交换机几种类型如下Fanout:广播将消息交给所有绑定到交换机的队列Direct:定向把消息交给符合指定routing key的队列Topic:通配符把消息交给符合routing pattern路由模式的队列生产者代码public class producer_PubSub { public static void main(String[] args) throws IOException, TimeoutException { //1.创建链接工厂 ConnectionFactory factory new ConnectionFactory(); //2.设置参数 factory.setHost(127.0.0.1);//ip地址 factory.setPort(5672); factory.setVirtualHost(/idcast);//虚拟机 factory.setUsername(yang); factory.setPassword(yang); //3.创建连接 Connection Connection connection factory.newConnection(); //4.创建channel Channel channel connection.createChannel(); //5.创建交换机 /** * exchange:交换机名称 * type:交换机类型 * durable:是否持久化 * autoDelete:自动删除 * internal:内部使用一般是false * arguments:参数 */ String exchangeName test_fanout; channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT,true,false,false,null); //6.创建队列 String queue1Name test_fanout_queue1; String queue2Name test_fanout_queue2; channel.queueDeclare(queue1Name,true,false,false,null); channel.queueDeclare(queue2Name,true,false,false,null); //7.绑定队列和交换机 /** * queue:队列名称 * exchange:交换机名称 * routingKey:路由键绑定规则 * 如果交换机的类型是fanout,routingKey设置为“” */ channel.queueBind(queue1Name,exchangeName,); channel.queueBind(queue2Name,exchangeName,); //8.发送消息 String body 日志信息; channel.basicPublish(exchangeName,,null,body.getBytes()); //9.释放资源 channel.close(); connection.close(); } }相比较工作模式这里多了创建交换机绑定交换机和队列这两步消费者代码public class comsume_PubSub1 { public static void main(String[] args) throws IOException, TimeoutException { //1.创建链接工厂 ConnectionFactory factory new ConnectionFactory(); //2.设置参数 factory.setHost(127.0.0.1);//ip地址 factory.setPort(5672); factory.setVirtualHost(/idcast);//虚拟机 factory.setUsername(yang); factory.setPassword(yang); //3.创建连接 Connection Connection connection factory.newConnection(); //4.创建channel Channel channel connection.createChannel(); //5.创建队列queue String queue1Name test_fanout_queue1; String queue2Name test_fanout_queue2; //6.接收消息 /** * queue:队列名称 * autoAck:是否自动确认 * callback:回调对象 */ Consumer comsumer new DefaultConsumer(channel){ Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { super.handleDelivery(consumerTag, envelope, properties, body); // System.out.println(consumerTag:consumerTag); // System.out.println(envelope:envelope.getExchange()); // System.out.println(envelope:envelope.getRoutingKey()); // System.out.println(properties:properties); System.out.println(body:new String(body)); System.out.println(将日志打印); } }; channel.basicConsume(queue1Name,true,comsumer); } }public class comsume_PubSub2 { public static void main(String[] args) throws IOException, TimeoutException { //1.创建链接工厂 ConnectionFactory factory new ConnectionFactory(); //2.设置参数 factory.setHost(127.0.0.1);//ip地址 factory.setPort(5672); factory.setVirtualHost(/idcast);//虚拟机 factory.setUsername(yang); factory.setPassword(yang); //3.创建连接 Connection Connection connection factory.newConnection(); //4.创建channel Channel channel connection.createChannel(); //5.创建队列queue String queue1Name test_fanout_queue1; String queue2Name test_fanout_queue2; //6.接收消息 /** * queue:队列名称 * autoAck:是否自动确认 * callback:回调对象 */ Consumer comsumer new DefaultConsumer(channel){ Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { super.handleDelivery(consumerTag, envelope, properties, body); // System.out.println(consumerTag:consumerTag); // System.out.println(envelope:envelope.getExchange()); // System.out.println(envelope:envelope.getRoutingKey()); // System.out.println(properties:properties); System.out.println(body:new String(body)); System.out.println(将日志打印保存数据库); } }; channel.basicConsume(queue2Name,true,comsumer); } }在这里这两个消费者都可以读到数据不是轮询的四.路由模式结构生产者发送消息时指定一个Routing Key。队列绑定到交换机时也指定一个或多个 Routing Key。只有当消息的 Routing Key 与队列的绑定 Key完全匹配时消息才会被路由到该队列。核心特点实现了定向精准投递。例如级别为error的日志只会发给专门处理错误的队列而不会被信息队列接收。注意1.队列与交换机绑定不是任意绑定了而是要所指定一个routingkey2.消息的发送方在向exchange发送消息的时候也必须指定routingkey3.exchange不在把消息交给每一个绑定的队列而是根据routingkey进行判断只有队列的routingkey和消息的routingkey一样是才会接收到消息生产者代码public class producer_Routing { public static void main(String[] args) throws IOException, TimeoutException { //1.创建链接工厂 ConnectionFactory factory new ConnectionFactory(); //2.设置参数 factory.setHost(127.0.0.1);//ip地址 factory.setPort(5672); factory.setVirtualHost(/idcast);//虚拟机 factory.setUsername(yang); factory.setPassword(yang); //3.创建连接 Connection Connection connection factory.newConnection(); //4.创建channel Channel channel connection.createChannel(); //5.创建交换机 /** * exchange:交换机名称 * type:交换机类型 * durable:是否持久化 * autoDelete:自动删除 * internal:内部使用一般是false * arguments:参数 */ String exchangeName test_direct; channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT,true,false,false,null); //6.创建队列 String queue1Name test_direct_queue1; String queue2Name test_direct_queue2; channel.queueDeclare(queue1Name,true,false,false,null); channel.queueDeclare(queue2Name,true,false,false,null); //7.绑定队列和交换机 /** * queue:队列名称 * exchange:交换机名称 * routingKey:路由键绑定规则 * 如果交换机的类型是fanout,routingKey设置为“” */ channel.queueBind(queue1Name,exchangeName,error); channel.queueBind(queue2Name,exchangeName,info); channel.queueBind(queue2Name,exchangeName,error); channel.queueBind(queue2Name,exchangeName,waring); //8.发送消息 String body 日志信息; channel.basicPublish(exchangeName,info,null,body.getBytes()); //9.释放资源 channel.close(); connection.close(); } }这里与发布订阅模式不同在这个模式写明了routingkey的要求消费者代码public class comsume_Rouring1 { public static void main(String[] args) throws IOException, TimeoutException { //1.创建链接工厂 ConnectionFactory factory new ConnectionFactory(); //2.设置参数 factory.setHost(127.0.0.1);//ip地址 factory.setPort(5672); factory.setVirtualHost(/idcast);//虚拟机 factory.setUsername(yang); factory.setPassword(yang); //3.创建连接 Connection Connection connection factory.newConnection(); //4.创建channel Channel channel connection.createChannel(); //5.创建队列queue String queue1Name test_direct_queue1; String queue2Name test_direct_queue2; //6.接收消息 /** * queue:队列名称 * autoAck:是否自动确认 * callback:回调对象 */ Consumer comsumer new DefaultConsumer(channel){ Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { super.handleDelivery(consumerTag, envelope, properties, body); // System.out.println(consumerTag:consumerTag); // System.out.println(envelope:envelope.getExchange()); // System.out.println(envelope:envelope.getRoutingKey()); // System.out.println(properties:properties); System.out.println(body:new String(body)); System.out.println(将日志打印); } }; channel.basicConsume(queue1Name,true,comsumer); } }public class comsume_Rouring2 { public static void main(String[] args) throws IOException, TimeoutException { //1.创建链接工厂 ConnectionFactory factory new ConnectionFactory(); //2.设置参数 factory.setHost(127.0.0.1);//ip地址 factory.setPort(5672); factory.setVirtualHost(/idcast);//虚拟机 factory.setUsername(yang); factory.setPassword(yang); //3.创建连接 Connection Connection connection factory.newConnection(); //4.创建channel Channel channel connection.createChannel(); //5.创建队列queue String queue1Name test_direct_queue1; String queue2Name test_direct_queue2; //6.接收消息 /** * queue:队列名称 * autoAck:是否自动确认 * callback:回调对象 */ Consumer comsumer new DefaultConsumer(channel){ Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { super.handleDelivery(consumerTag, envelope, properties, body); // System.out.println(consumerTag:consumerTag); // System.out.println(envelope:envelope.getExchange()); // System.out.println(envelope:envelope.getRoutingKey()); // System.out.println(properties:properties); System.out.println(body:new String(body)); System.out.println(将日志打印到控制台); } }; channel.basicConsume(queue2Name,true,comsumer); } } 注意routingkey要与对应的生产者的routingkey相一致五.主题模式结构与路由模式类似但 Routing Key 支持通配符*匹配一个单词。#匹配零个或多个单词。核心特点实现了灵活的一对多路由。例如绑定log.*.error可以匹配log.order.error和log.user.error绑定log.#则可以匹配log.info、log.order.error等任意以 log 开头的消息。生产者代码:public class producer_Topic { public static void main(String[] args) throws IOException, TimeoutException { //1.创建链接工厂 ConnectionFactory factory new ConnectionFactory(); //2.设置参数 factory.setHost(127.0.0.1);//ip地址 factory.setPort(5672); factory.setVirtualHost(/idcast);//虚拟机 factory.setUsername(yang); factory.setPassword(yang); //3.创建连接 Connection Connection connection factory.newConnection(); //4.创建channel Channel channel connection.createChannel(); //5.创建交换机 /** * exchange:交换机名称 * type:交换机类型 * durable:是否持久化 * autoDelete:自动删除 * internal:内部使用一般是false * arguments:参数 */ String exchangeName test_topic; channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC,true,false,false,null); //6.创建队列 String queue1Name test_topic1; String queue2Name test_topic2; channel.queueDeclare(queue1Name,true,false,false,null); channel.queueDeclare(queue2Name,true,false,false,null); //7.绑定队列和交换机 /** * queue:队列名称 * exchange:交换机名称 * routingKey:路由键绑定规则 * 如果交换机的类型是fanout,routingKey设置为“” */ channel.queueBind(queue1Name,exchangeName,#.error); channel.queueBind(queue1Name,exchangeName,order.*); channel.queueBind(queue2Name,exchangeName,*.*); //8.发送消息 String body 日志信息; channel.basicPublish(exchangeName,order.info,null,body.getBytes()); //9.释放资源 channel.close(); connection.close(); } }消费者代码public class comsume_Topic1 { public static void main(String[] args) throws IOException, TimeoutException { //1.创建链接工厂 ConnectionFactory factory new ConnectionFactory(); //2.设置参数 factory.setHost(127.0.0.1);//ip地址 factory.setPort(5672); factory.setVirtualHost(/idcast);//虚拟机 factory.setUsername(yang); factory.setPassword(yang); //3.创建连接 Connection Connection connection factory.newConnection(); //4.创建channel Channel channel connection.createChannel(); //5.创建队列queue String queue1Name test_topic1; String queue2Name test_topic2; //6.接收消息 /** * queue:队列名称 * autoAck:是否自动确认 * callback:回调对象 */ Consumer comsumer new DefaultConsumer(channel){ Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { super.handleDelivery(consumerTag, envelope, properties, body); // System.out.println(consumerTag:consumerTag); // System.out.println(envelope:envelope.getExchange()); // System.out.println(envelope:envelope.getRoutingKey()); // System.out.println(properties:properties); System.out.println(body:new String(body)); System.out.println(将日志打印); } }; channel.basicConsume(queue1Name,true,comsumer); } }public class comsume_Topic2 { public static void main(String[] args) throws IOException, TimeoutException { //1.创建链接工厂 ConnectionFactory factory new ConnectionFactory(); //2.设置参数 factory.setHost(127.0.0.1);//ip地址 factory.setPort(5672); factory.setVirtualHost(/idcast);//虚拟机 factory.setUsername(yang); factory.setPassword(yang); //3.创建连接 Connection Connection connection factory.newConnection(); //4.创建channel Channel channel connection.createChannel(); //5.创建队列queue String queue1Name test_topic1; String queue2Name test_topic2; //6.接收消息 /** * queue:队列名称 * autoAck:是否自动确认 * callback:回调对象 */ Consumer comsumer new DefaultConsumer(channel){ Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { super.handleDelivery(consumerTag, envelope, properties, body); // System.out.println(consumerTag:consumerTag); // System.out.println(envelope:envelope.getExchange()); // System.out.println(envelope:envelope.getRoutingKey()); // System.out.println(properties:properties); System.out.println(body:new String(body)); System.out.println(将日志打印控制台); } }; channel.basicConsume(queue2Name,true,comsumer); } }在配置routingkey的时候可以使用通配符更加灵活