七种模式介绍与利用场景
简略模式（Hello World）
做最简略的事件，一个生产者对应一个消费者，RabbitMQ相当于一个音讯代理，负责将A的音讯转发给B
利用场景：将发送的电子邮件放到音讯队列，而后邮件服务在队列中获取邮件并发送给收件人
工作队列模式（Work queues）
在多个消费者之间分配任务（竞争的消费者模式），一个生产者对应多个消费者，个别实用于执行资源密集型工作，单个消费者解决不过去，须要多个消费者进行解决
利用场景：一个订单的解决须要10s，有多个订单能够同时放到音讯队列，而后让多个消费者同时解决，这样就是并行了，而不是单个消费者的串行状况
订阅模式（Publish/Subscribe）
一次向许多消费者发送音讯，一个生产者发送的音讯会被多个消费者获取，也就是将音讯将播送到所有的消费者中。
利用场景：更新商品库存后须要告诉多个缓存和多个数据库，这里的构造应该是：
- 一个fanout类型交换机扇出两个个音讯队列，别离为缓存音讯队列、数据库音讯队列
- 一个缓存音讯队列对应着多个缓存消费者
- 一个数据库音讯队列对应着多个数据库消费者
路由模式（Routing）
有选择地（Routing key）接管音讯，发送音讯到交换机并且要指定路由key ，消费者将队列绑定到交换机时须要指定路由key，仅生产指定路由key的音讯
利用场景：如在商品库存中减少了1台iphone12，iphone12促销流动消费者指定routing key为iphone12，只有此促销流动会接管到音讯，其它促销流动不关怀也不会生产此routing key的音讯
主题模式（Topics）
依据主题（Topics）来接管音讯，将路由key和某模式进行匹配，此时队列须要绑定在一个模式上，#匹配一个词或多个词，*只匹配一个词。
利用场景： 同上，iphone促销流动能够接管主题为iphone的音讯，如iphone12、iphone13等
近程过程调用（RPC）
如果咱们须要在近程计算机上运行性能并期待后果就能够应用RPC，具体流程能够看图。利用场景：须要期待接口返回数据，如订单领取
发布者确认（Publisher Confirms）
与发布者进行牢靠的公布确认，发布者确认是RabbitMQ扩大，能够实现牢靠的公布。在通道上启用发布者确认后，RabbitMQ将异步确认发送者公布的音讯，这意味着它们已在服务器端解决。（搜寻公众号民工哥技术之路，查看更多 Java后端技术栈及面试题精选）
利用场景：对于音讯可靠性要求较高，比方钱包扣款
代码演示
代码中没有对前面两种模式演示，有趣味能够本人钻研
简略模式
import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Sender { private final static String QUEUE_NAME = "simple_queue"; public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); // 申明队列 // queue：队列名 // durable：是否长久化 // exclusive：是否排外 即只容许该channel拜访该队列 个别等于true的话用于一个队列只能有一个消费者来生产的场景 // autoDelete：是否主动删除 生产完删除 // arguments：其余属性 channel.queueDeclare(QUEUE_NAME, false, false, false, null); //音讯内容 String message = "simplest mode message"; channel.basicPublish("", QUEUE_NAME, null, message.getBytes()); System.out.println("[x]Sent '" + message + "'"); //最初敞开通关和连贯 channel.close(); connection.close(); }}import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import com.rabbitmq.client.DeliverCallback;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Receiver { private final static String QUEUE_NAME = "simplest_queue"; public static void main(String[] args) throws IOException, InterruptedException, TimeoutException { // 获取连贯 ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.queueDeclare(QUEUE_NAME, false, false, false, null); DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody(), "UTF-8"); System.out.println(" [x] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'"); }; channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { }); }}工作队列模式
import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import com.rabbitmq.client.DeliverCallback;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Receiver1 { private final static String QUEUE_NAME = "queue_work"; public static void main(String[] args) throws IOException, InterruptedException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 同一时刻服务器只会发送一条音讯给消费者 channel.basicQos(1); DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody(), "UTF-8"); System.out.println(" [x] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'"); }; channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { }); }}import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import com.rabbitmq.client.DeliverCallback;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Receiver2 { private final static String QUEUE_NAME = "queue_work"; public static void main(String[] args) throws IOException, InterruptedException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 同一时刻服务器只会发送一条音讯给消费者 channel.basicQos(1); DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody(), "UTF-8"); System.out.println(" [x] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'"); }; channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { }); }}import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Sender { private final static String QUEUE_NAME = "queue_work"; public static void main(String[] args) throws IOException, InterruptedException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); // 申明队列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); for (int i = 0; i < 100; i++) { String message = "work mode message" + i; channel.basicPublish("", QUEUE_NAME, null, message.getBytes()); System.out.println("[x] Sent '" + message + "'"); Thread.sleep(i * 10); } channel.close(); connection.close(); }}公布订阅模式
import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import com.rabbitmq.client.DeliverCallback;public class Receive1 { private static final String EXCHANGE_NAME = "logs"; public static void main(String[] argv) throws Exception { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.exchangeDeclare(EXCHANGE_NAME, "fanout"); String queueName = channel.queueDeclare().getQueue(); channel.queueBind(queueName, EXCHANGE_NAME, ""); System.out.println(" [*] Waiting for messages. To exit press CTRL+C"); // 订阅音讯的回调函数 DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody(), "UTF-8"); System.out.println(" [x] Received '" + message + "'"); }; // 消费者，有音讯时登程订阅回调函数 channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { }); }}import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import com.rabbitmq.client.DeliverCallback;public class Receive2 { private static final String EXCHANGE_NAME = "logs"; public static void main(String[] argv) throws Exception { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.exchangeDeclare(EXCHANGE_NAME, "fanout"); String queueName = channel.queueDeclare().getQueue(); channel.queueBind(queueName, EXCHANGE_NAME, ""); System.out.println(" [*] Waiting for messages. To exit press CTRL+C"); // 订阅音讯的回调函数 DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody(), "UTF-8"); System.out.println(" [x] Received2 '" + message + "'"); }; // 消费者，有音讯时登程订阅回调函数 channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { }); }}import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;public class Sender { private static final String EXCHANGE_NAME = "logs"; public static void main(String[] argv) throws Exception { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.exchangeDeclare(EXCHANGE_NAME, "fanout"); String message = "publish subscribe message"; channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8")); System.out.println(" [x] Sent '" + message + "'"); channel.close(); connection.close(); }}路由模式
import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import com.rabbitmq.client.DeliverCallback;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Receiver1 { private final static String QUEUE_NAME = "queue_routing"; private final static String EXCHANGE_NAME = "exchange_direct"; public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 指定路由的key，接管key和key2 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key"); channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key2"); channel.basicQos(1); DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody(), "UTF-8"); System.out.println(" [x] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'"); }; channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { }); }}import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import com.rabbitmq.client.DeliverCallback;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Receiver2 { private final static String QUEUE_NAME = "queue_routing2"; private final static String EXCHANGE_NAME = "exchange_direct"; public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 仅接管key2 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key2"); channel.basicQos(1); DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody(), "UTF-8"); System.out.println(" [x] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'"); }; channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { }); }}import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Sender { private final static String EXCHANGE_NAME = "exchange_direct"; private final static String EXCHANGE_TYPE = "direct"; public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); // 交换机申明 channel.exchangeDeclare(EXCHANGE_NAME, EXCHANGE_TYPE); // 只有routingKey雷同的才会生产 String message = "routing mode message"; channel.basicPublish(EXCHANGE_NAME, "key2", null, message.getBytes()); System.out.println("[x] Sent '" + message + "'");// channel.basicPublish(EXCHANGE_NAME, "key", null, message.getBytes());// System.out.println("[x] Sent '" + message + "'"); channel.close(); connection.close(); }}主题模式
import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import com.rabbitmq.client.DeliverCallback;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Receiver1 { private final static String QUEUE_NAME = "queue_topic"; private final static String EXCHANGE_NAME = "exchange_topic"; public static void main(String[] args) throws IOException, InterruptedException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 能够接管key.1 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key.*"); channel.basicQos(1); DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody(), "UTF-8"); System.out.println(" [x] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'"); }; channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { }); }}import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import com.rabbitmq.client.DeliverCallback;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Receiver2 { private final static String QUEUE_NAME = "queue_topic2"; private final static String EXCHANGE_NAME = "exchange_topic"; private final static String EXCHANGE_TYPE = "topic"; public static void main(String[] args) throws IOException, InterruptedException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.queueDeclare(QUEUE_NAME, false, false, false, null); // *号代表单个单词，能够接管key.1 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "*.*"); // #号代表多个单词，能够接管key.1.2 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "*.#"); channel.basicQos(1); DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody(), "UTF-8"); System.out.println(" [x] Received '" + delivery.getEnvelope().getRoutingKey() + "':'" + message + "'"); }; channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { }); }}import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Sender { private final static String EXCHANGE_NAME = "exchange_topic"; private final static String EXCHANGE_TYPE = "topic"; public static void main(String[] args) throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); factory.setPort(5672); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.exchangeDeclare(EXCHANGE_NAME, EXCHANGE_TYPE); String message = "topics model message with key.1"; channel.basicPublish(EXCHANGE_NAME, "key.1", null, message.getBytes()); System.out.println("[x] Sent '" + message + "'"); String message2 = "topics model message with key.1.2"; channel.basicPublish(EXCHANGE_NAME, "key.1.2", null, message2.getBytes()); System.out.println("[x] Sent '" + message2 + "'"); channel.close(); connection.close(); }}四种交换机介绍
- 直连交换机（Direct exchange）： 具备路由性能的交换机，绑定到此交换机的时候须要指定一个routing_key，交换机发送音讯的时候须要routing_key，会将音讯发送道对应的队列
- 扇形交换机（Fanout exchange）： 播送音讯到所有队列，没有任何解决，速度最快
- 主题交换机（Topic exchange）： 在直连交换机根底上减少模式匹配，也就是对routing_key进行模式匹配，*代表一个单词，#代表多个单词
- 首部交换机（Headers exchange）： 疏忽routing_key，应用Headers信息（一个Hash的数据结构）进行匹配，劣势在于能够有更多更灵便的匹配规定
总结
这么多种队列模式中都有其利用场景，大家能够依据利用场景示例中进行抉择
作者：我思知我在
blog.csdn.net/qq_32828253/article/details/110450249