消息中间件之RabbitMQ_vlambda技术博客

vlambda
2021-05-06

消息中间件之RabbitMQ

随着微服务大行其道，MQ的使用场景也跟着多了起来，就拿我们的电商系统来说:

某电商系统中包含有订单服务和商品服务，这两个是不同相互独立的微服务，系统B操作成功的时候，系统A需要进行库存的扣减操作。按照以往，订单这块服务，需要调用商品服务进行相关操作，成功后才返回下单成功。这显然违背了独立这两个字，增加了系统之间的耦合性，还有当网络发生抖动时，会致使用户体验非常之差，再者高并发的情况下，这是非常要不得的。这时我们的MQ就派上了用场。消息队列(MQ)提供一个异步通信机制，消息的发送者不必一直等待到消息被成功处理才返回，而是立即返回。消息中间件负责处理网络通信，如果网络连接不可用，消息被暂存于队列当中，当网络畅通的时候在将消息转发给相应的应用程序或者服务，当然前提是这些服务订阅了该队列。

消息中间件之RabbitMQ

下订单后，订单系统给消息中间件发布一个扣减该商品的消息进而返回完成订单，至于商品系统有无扣减这就不关订单系统的事了。那么，问题来了，很多童鞋可能会问，那岂不是会错乱了，万一扣减不成功呢。当然了这确实是个问题，但这又是另一个话题了，那就是一致性，在这里不讨论，日后再抽出个主题讲解一下。话说回来，当商品系统通过消息中间件接到这个消息后，进行相应的扣减操作。如果在商品服务和订单服务之间使用消息中间件，既可以提高并发量，又降低服务之间的耦合度。何乐而不为呢！

关于消息中间件呢，有很多，而RabbitMQ就是其中之一，我们该如何玩转它呢，跟着我的步骤一起嗨起来：

1.安装，不安装哪来的使用啊

windows之下的安装:

需要准备的安装包:

1）RabbitMQ基于Erlang语言,需要依赖Erlang，所以需要先安装Erlang：

https://www.erlang.org/downloads

2）RabbitMQ安装包:

http://www.rabbitmq.com/download.html

至于具体怎么安装，很简单，像其他软件一下双击安装包一步步安装就好了，在这里不多说。

Linux下的安装(以CentOS8为例):

需要准备的包:

1）Erlang rpm安装包

https://github.com/rabbitmq/erlang-rpm/releases

2) socat rpm安装包

https://pkgs.org/download/socat

3) rabbitmq-server rpm安装包

https://www.rabbitmq.com/download.html

下载相应的包要注意哦，如下图:

消息中间件之RabbitMQ

Erlang版本一定要对上rabbitmq-server版本，不然够你折腾了，掉坑了还不知道怎么爬上来。

下载好相关包后:

运行命令yum install erlang-23.3.1-1.el8.x86_64.rpm先安装Erlang

然后运行yum install socat-2.0.0-0.b9.9.mga8.x86_64.rpm 安装socat

最后运行yum install rabbitmq-server-3.8.9-1.el8.noarch.rpm 安装rabbitmq

都安装完成后，需要以下步骤:

1)启动rabbitmq服务：service rabbitmq-server start

2)启动插件页面管理：rabbitmq-plugins enable rabbitmq_management

3)创建用户(用户名密码)：rabbitmqctl add_user admin admin

4)为用户添加角色：rabbitmqctl set_user_tags admin administrator

5)赋予权限：rabbitmqctl set_permissions -p / admin ".*" ".*" ".*"

然后就可以通过浏览器访问控制台了(管理控制台端口号默认是：15672)

消息中间件之RabbitMQ

登录成功后如下

消息中间件之RabbitMQ

RabbitMQ的7种队列模式：

1.简单模式(Hello World)

消息中间件之RabbitMQ

只有一个生产者和一个消费者，RabbitMQ相当于一个消息代理，负责将P的消息转发给C。
应用场景：将发送的电子邮件放到消息队列，然后邮件服务在队列中获取邮件并发送给收件人

2.工作队列模式(Work Queues)

消息中间件之RabbitMQ

在多个消费者之间分配任务（竞争的消费者模式），一个生产者对应多个消费者，一般适用于执行资源密集型任务，单个消费者处理不过来，需要多个消费者进行处理。

应用场景：一个订单的处理需要10s，有多个订单可以同时放到消息队列，然后让多个消费者同时处理，这样就是并行了，而不是单个消费者的串行情况

3.发布订阅模式(Public/Subscribe)

消息中间件之RabbitMQ

一次向许多消费者发送消息，一个生产者发送的消息会被多个消费者获取，也就是将消息将广播到所有的消费者中。

应用场景：更新商品库存后需要通知多个缓存和多个数据库，这里的结构应该是：

①一个fanout类型交换机扇出两个个消息队列，分别为缓存消息队列、数据库消息队列

②一个缓存消息队列对应着多个缓存消费者

③一个数据库消息队列对应着多个数据库消费者

4.路由模式(Routing)

消息中间件之RabbitMQ

有选择地（Routing key）接收消息，发送消息到交换机并且要指定路由key ，消费者将队列绑定到交换机时需要指定路由key，仅消费指定路由key的消息

应用场景：如在商品库存中增加了1台iphone14，iphone14促销活动消费者指定routing key为iphone14，只有此促销活动会接收到消息，其它促销活动不关心也不会消费此routing key的消息

5.主题模式(Topics)

消息中间件之RabbitMQ

根据主题(Topics)来接收消息，将路由key和某模式进行匹配，此时队列需要绑定在一个模式上，#匹配一个词或多个词，*只匹配一个词。
应用场景：同上，iphone促销活动可以接收主题为iphone的消息，如iphone12、iphone13等

6.远程过程调用(RPC)

如果我们需要在远程计算机上运行功能并等待结果就可以使用RPC，具体流程可以看图。
应用场景：需要等待接口返回数据，如订单支付

7.发布者确认(Publisher Confirms)

与发布者进行可靠的确认，发布者确认是RabbitMQ扩展，可以实现可靠的发布。在通道上启用发布者确认后，RabbitMQ将异步确认发送者发布的消息，这意味着它们已在服务器端处理
应用场景：对于消息可靠性要求较高，比如钱包扣款

这几种模式，且看如下代码：

以下项目需要再项目中引入相关的jar包，如maven：

<dependency>  <groupId>com.rabbitmq</groupId>  <artifactId>amqp-client</artifactId>  <version>5.10.0</version></dependency>

简单模式

先登录管理控制台建立一个名为mq_test_01的Virtual host

生产者：

import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import java.nio.charset.StandardCharsets;
public class Send {    private final static String QUEUE_NAME = "mq_test_01"; public static void main(String[] argv) throws Exception { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) {             /**               *参数1 队列名               *参数2 是否持久化               *参数3 是否允许该channel访问该队列，为true则一个队列只能有一个消费者来消费               *参数4 是否自动删除，即消费完就删除               *参数5 其他属性               **/ channel.queueDeclare(QUEUE_NAME, false, false, false, null);            //要发送的消息 String message = "Hello World!"; channel.basicPublish("", QUEUE_NAME, null, message.getBytes(StandardCharsets.UTF_8)); System.out.println(" [x] Sent '" + message + "'");            //因为Channel和Connection继承自Closeable，所以不用声明关闭 } }}

消费者：

import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import com.rabbitmq.client.DeliverCallback;
public class Recv {
    private final static String QUEUE_NAME = "mq_test_01";
 public static void main(String[] argv) throws Exception { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel();
 channel.queueDeclare(QUEUE_NAME, false, false, false, null); 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(QUEUE_NAME, true, deliverCallback, consumerTag -> { }); }}

工作队列模式

生产者

import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import com.rabbitmq.client.MessageProperties;public class NewTask{    private static final String TASK_QUEUE_NAME = "mq_test_02"; public static void main(String[] argv) throws Exception { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) {            channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);            String message = String.join(" ", argv); channel.basicPublish("", TASK_QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes("UTF-8")); System.out.println(" [x] Sent '" + message + "'"); }    }}

消费者1

import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import com.rabbitmq.client.DeliverCallback;
public class Worker1{
 private static final String TASK_QUEUE_NAME = "task_queue";
 public static void main(String[] argv) throws Exception { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); final Connection connection = factory.newConnection(); final Channel channel = connection.createChannel();
 channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null); System.out.println(" [*] Waiting for messages. To exit press CTRL+C");        //该消费者在接收到队列里的消息但没有返回确认结果之前,rabbitmq不会将新的消息分发给该消费者 channel.basicQos(1);
 DeliverCallback deliverCallback = (consumerTag, delivery) -> {            String message = new String(delivery.getBody(), "UTF-8"); System.out.println(" [x] Received '" + message + "'"); try { doWork(message); } finally { System.out.println(" [x] Done"); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } }; channel.basicConsume(TASK_QUEUE_NAME, false, deliverCallback, consumerTag -> { });    } private static void doWork(String task) { for (char ch : task.toCharArray()) { if (ch == '.') { try { Thread.sleep(1000); } catch (InterruptedException _ignored) { Thread.currentThread().interrupt(); } } } }}

消费者2

import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import com.rabbitmq.client.DeliverCallback;
public class Worker2{
 private static final String TASK_QUEUE_NAME = "task_queue";
 public static void main(String[] argv) throws Exception { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); final Connection connection = factory.newConnection(); final Channel channel = connection.createChannel();
 channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null); System.out.println(" [*] Waiting for messages. To exit press CTRL+C"); //该消费者在接收到队列里的消息但没有返回确认结果之前,rabbitmq不会将新的消息分发给该消费者 channel.basicQos(1);
 DeliverCallback deliverCallback = (consumerTag, delivery) -> { String message = new String(delivery.getBody(), "UTF-8"); System.out.println(" [x] Received '" + message + "'"); try { doWork(message); } finally { System.out.println(" [x] Done"); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); } }; channel.basicConsume(TASK_QUEUE_NAME, false, deliverCallback, consumerTag -> { }); } private static void doWork(String task) { for (char ch : task.toCharArray()) { if (ch == '.') { try { Thread.sleep(1000); } catch (InterruptedException _ignored) { Thread.currentThread().interrupt(); } } } }}

发布订阅模式

发布者：

import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;
public class EmitLog {    private static final String EXCHANGE_NAME = "mq_test_01"; public static void main(String[] argv) throws Exception { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) { channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
 String message = argv.length < 1 ? "info: Hello World!" : String.join(" ", argv);
 channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8")); System.out.println(" [x] Sent '" + message + "'"); }    }}

订阅者1：

import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import com.rabbitmq.client.DeliverCallback;
public class ReceiveLogs1 {    private static final String EXCHANGE_NAME = "mq_test_01";
 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 -> { }); }}

订阅者2

import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import com.rabbitmq.client.DeliverCallback;
public class ReceiveLogs2 {    private static final String EXCHANGE_NAME = "mq_test_01";
 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;
public class EmitLogDirect {
  private static final String EXCHANGE_NAME = "mq_test_01";
 public static void main(String[] argv) throws Exception { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) { //交换机声明        channel.exchangeDeclare(EXCHANGE_NAME, "direct"); String message = "message";        //只有只有 参数2 相同才会接收消息        channel.basicPublish(EXCHANGE_NAME, "key1", null, message.getBytes("UTF-8")); System.out.println(" [x] Sent " + message + "'"); }  }}

消费者1

import com.rabbitmq.client.*;
public class ReceiveLogsDirect1 {  private static final String EXCHANGE_NAME = "mq_test_01";  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, "direct"); String queueName = channel.queueDeclare().getQueue(); channel.queueDeclare(EXCHANGE_NAME, false, false, false, null);     // 仅接收key2     channel.queueBind(queueName, 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(queueName, true, deliverCallback, consumerTag -> {      }); }}

消费者2：

import com.rabbitmq.client.*;
public class ReceiveLogsDirect2 { private static final String EXCHANGE_NAME = "mq_test_01";
 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, "direct"); String queueName = channel.queueDeclare().getQueue(); channel.queueDeclare(QUEUE_NAME, false, false, false, null);     // 仅接收key1和key2     channel.queueBind(queueName, EXCHANGE_NAME, "key1") channel.queueBind(queueName, 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(queueName, 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 + "'"); }}

消费者1

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 -> { }); }}

消费者2

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 -> { }); }}

RPC模式

RPCServer：

import com.rabbitmq.client.*;
public class RPCServer {
 private static final String RPC_QUEUE_NAME = "rpc_queue";
 private static int fib(int n) { if (n == 0) return 0; if (n == 1) return 1; return fib(n - 1) + fib(n - 2); }
 public static void main(String[] argv) throws Exception { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost");
 try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) { channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null); channel.queuePurge(RPC_QUEUE_NAME);
 channel.basicQos(1);
 System.out.println(" [x] Awaiting RPC requests");
 Object monitor = new Object(); DeliverCallback deliverCallback = (consumerTag, delivery) -> { AMQP.BasicProperties replyProps = new AMQP.BasicProperties .Builder() .correlationId(delivery.getProperties().getCorrelationId()) .build();
 String response = "";
 try { String message = new String(delivery.getBody(), "UTF-8"); int n = Integer.parseInt(message);
 System.out.println(" [.] fib(" + message + ")"); response += fib(n); } catch (RuntimeException e) { System.out.println(" [.] " + e.toString()); } finally { channel.basicPublish("", delivery.getProperties().getReplyTo(), replyProps, response.getBytes("UTF-8")); channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);                    // RabbitMq 消费者工作线程通知 RPC 服务拥有者线程 synchronized (monitor) { monitor.notify(); } } };
 channel.basicConsume(RPC_QUEUE_NAME, false, deliverCallback, (consumerTag -> { }));            //等待并准备使用来自RPC客户机的消息 while (true) { synchronized (monitor) { try { monitor.wait(); } catch (InterruptedException e) { e.printStackTrace(); } } } } }}

RPC Client：

import com.rabbitmq.client.AMQP;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;import java.util.UUID;import java.util.concurrent.ArrayBlockingQueue;import java.util.concurrent.BlockingQueue;import java.util.concurrent.TimeoutException;
public class RPCClient implements AutoCloseable {
 private Connection connection; private Channel channel; private String requestQueueName = "rpc_queue";
 public RPCClient() throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost");
 connection = factory.newConnection(); channel = connection.createChannel(); }
 public static void main(String[] argv) { try (RPCClient fibonacciRpc = new RPCClient()) { for (int i = 0; i < 32; i++) { String i_str = Integer.toString(i); System.out.println(" [x] Requesting fib(" + i_str + ")"); String response = fibonacciRpc.call(i_str); System.out.println(" [.] Got '" + response + "'"); } } catch (IOException | TimeoutException | InterruptedException e) { e.printStackTrace(); } }
 public String call(String message) throws IOException, InterruptedException { final String corrId = UUID.randomUUID().toString();
 String replyQueueName = channel.queueDeclare().getQueue(); AMQP.BasicProperties props = new AMQP.BasicProperties .Builder() .correlationId(corrId) .replyTo(replyQueueName) .build();
 channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8"));
 final BlockingQueue<String> response = new ArrayBlockingQueue<>(1);
 String ctag = channel.basicConsume(replyQueueName, true, (consumerTag, delivery) -> { if (delivery.getProperties().getCorrelationId().equals(corrId)) { response.offer(new String(delivery.getBody(), "UTF-8")); } }, consumerTag -> {        }); String result = response.take(); channel.basicCancel(ctag); return result;    } public void close() throws IOException { connection.close(); }}

发布者确认模式

import com.rabbitmq.client.Channel;import com.rabbitmq.client.ConfirmCallback;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;
import java.time.Duration;import java.util.UUID;import java.util.concurrent.ConcurrentNavigableMap;import java.util.concurrent.ConcurrentSkipListMap;import java.util.function.BooleanSupplier;
public class PublisherConfirms {
 static final int MESSAGE_COUNT = 50_000;
 static Connection createConnection() throws Exception { ConnectionFactory cf = new ConnectionFactory(); cf.setHost("localhost"); cf.setUsername("guest"); cf.setPassword("guest"); return cf.newConnection(); }
 public static void main(String[] args) throws Exception { publishMessagesIndividually(); publishMessagesInBatch(); handlePublishConfirmsAsynchronously(); }
 static void publishMessagesIndividually() throws Exception { try (Connection connection = createConnection()) { Channel ch = connection.createChannel();
 String queue = UUID.randomUUID().toString(); ch.queueDeclare(queue, false, false, true, null);
 ch.confirmSelect(); long start = System.nanoTime(); for (int i = 0; i < MESSAGE_COUNT; i++) { String body = String.valueOf(i); ch.basicPublish("", queue, null, body.getBytes()); ch.waitForConfirmsOrDie(5_000); } long end = System.nanoTime(); System.out.format("Published %,d messages individually in %,d ms%n", MESSAGE_COUNT, Duration.ofNanos(end - start).toMillis()); } }
 static void publishMessagesInBatch() throws Exception { try (Connection connection = createConnection()) { Channel ch = connection.createChannel();
 String queue = UUID.randomUUID().toString(); ch.queueDeclare(queue, false, false, true, null);
 ch.confirmSelect();
 int batchSize = 100; int outstandingMessageCount = 0;
 long start = System.nanoTime(); for (int i = 0; i < MESSAGE_COUNT; i++) { String body = String.valueOf(i); ch.basicPublish("", queue, null, body.getBytes()); outstandingMessageCount++;
 if (outstandingMessageCount == batchSize) { ch.waitForConfirmsOrDie(5_000); outstandingMessageCount = 0; } }
 if (outstandingMessageCount > 0) { ch.waitForConfirmsOrDie(5_000); } long end = System.nanoTime(); System.out.format("Published %,d messages in batch in %,d ms%n", MESSAGE_COUNT, Duration.ofNanos(end - start).toMillis()); } }
 static void handlePublishConfirmsAsynchronously() throws Exception { try (Connection connection = createConnection()) { Channel ch = connection.createChannel();
 String queue = UUID.randomUUID().toString(); ch.queueDeclare(queue, false, false, true, null);
 ch.confirmSelect();
 ConcurrentNavigableMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();
 ConfirmCallback cleanOutstandingConfirms = (sequenceNumber, multiple) -> { if (multiple) { ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap( sequenceNumber, true ); confirmed.clear(); } else { outstandingConfirms.remove(sequenceNumber); } };
 ch.addConfirmListener(cleanOutstandingConfirms, (sequenceNumber, multiple) -> { String body = outstandingConfirms.get(sequenceNumber); System.err.format( "Message with body %s has been nack-ed. Sequence number: %d, multiple: %b%n", body, sequenceNumber, multiple ); cleanOutstandingConfirms.handle(sequenceNumber, multiple); });
 long start = System.nanoTime(); for (int i = 0; i < MESSAGE_COUNT; i++) { String body = String.valueOf(i); outstandingConfirms.put(ch.getNextPublishSeqNo(), body); ch.basicPublish("", queue, null, body.getBytes()); }
 if (!waitUntil(Duration.ofSeconds(60), () -> outstandingConfirms.isEmpty())) { throw new IllegalStateException("All messages could not be confirmed in 60 seconds"); }
 long end = System.nanoTime(); System.out.format("Published %,d messages and handled confirms asynchronously in %,d ms%n", MESSAGE_COUNT, Duration.ofNanos(end - start).toMillis()); } }
 static boolean waitUntil(Duration timeout, BooleanSupplier condition) throws InterruptedException { int waited = 0; while (!condition.getAsBoolean() && waited < timeout.toMillis()) { Thread.sleep(100L); waited = +100; } return condition.getAsBoolean(); }
}

好了，本篇到此完结

vlambda博客
学习文章列表