文章目录
- 六. RPC(RPC通信模式)
- 客户端
- 服务端
- 七. Publisher Confirms(发布确认模式)
- 1. Publishing Messages Individually(单独确认)
- 2. Publishing Messages in Batches(批量确认)
- 3. Handling Publisher Confirms Asynchronously(异步确认)
六. RPC(RPC通信模式)
- 客⼾端发送消息到⼀个指定的队列, 并在消息属性中设置replyTo字段, 这个字段指定了⼀个回调队列, ⽤于接收服务端的响应.
- 服务端接收到请求后, 处理请求并发送响应消息到replyTo指定的回调队列
- 客⼾端在回调队列上等待响应消息. ⼀旦收到响应，客⼾端会检查消息的correlationId属性，以确保它是所期望的响应
公共代码:
public static final String RPC_REQUEST_QUEUE = "rpc.request.queue";
public static final String RPC_RESPONSE_QUEUE = "rpc.response.queue";
客户端
客户端需要完成两件事, 发送请求, 接收响应
发送请求:
//发送请求:
//声明队列
channel.queueDeclare(Common.RPC_REQUEST_QUEUE, true, false, false, null);
channel.queueDeclare(Common.RPC_RESPONSE_QUEUE, true, false, false, null);
//定义回调队列
String replyQueueName = Common.RPC_RESPONSE_QUEUE;
//本次请求的唯一标识
String corrId = UUID.randomUUID().toString();
//生成发送消息的属性
AMQP.BasicProperties props = new AMQP.BasicProperties
.Builder()
.correlationId(corrId)
.replyTo(replyQueueName)
.build();
//通过内置交换机, 发送消息
String message = "hello, rpc......";
channel.basicPublish("", Common.RPC_REQUEST_QUEUE, props, message.getBytes(StandardCharsets.UTF_8));
接收响应:
//接收响应:
//使用阻塞队列来存储回调结果, 避免了客户端反复访问队列
final BlockingQueue<String> response = new ArrayBlockingQueue<>(1);
//接收服务器的响应
DefaultConsumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("接收到消息: " + new String(body));
//如果唯一标识正确, 放在阻塞队列中
if(properties.getCorrelationId().equals(corrId)){
response.offer(new String(body));
}
}
};
channel.basicConsume(replyQueueName, true, consumer);
//获取回调的结果
String result = response.take();
System.out.println("[RPCClient] Result: " + result);
服务端
//设置同时最多只能获取一个消息
channel.basicQos(1);
//接收消息, 并对消息进行应答
DefaultConsumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
BasicProperties replyProps = new AMQP.BasicProperties
.Builder()
.correlationId(properties.getCorrelationId())
.build();
String message = new String(body);
String response = "接收到消息 request: " + message;
channel.basicPublish("", properties.getReplyTo(), replyProps, response.getBytes(StandardCharsets.UTF_8));
//对消息进行应答
channel.basicAck(envelope.getDeliveryTag(), false);
}
};
channel.basicConsume(Common.RPC_REQUEST_QUEUE, false, consumer);//自动应答设置成false, 在成功回调后, 再进行手动应答
七. Publisher Confirms(发布确认模式)
Publisher Confirms模式是RabbitMQ提供的⼀种确保消息可靠发送到RabbitMQ服务器的机制。在这种模式下，⽣产者可以等待RabbitMQ服务器的确认，以确保消息已经被服务器接收并处理.
- ⽣产者将Channel设置为confirm模式(通过调⽤channel.confirmSelect()完成)后, 发布的每⼀条消息都会获得⼀个唯⼀的ID, ⽣产者可以将这些序列号与消息关联起来，以便跟踪消息的状态.
- 当消息被RabbitMQ服务器接收并处理后，服务器会异步地向⽣产者发送⼀个确认(ACK)给⽣产者(包含消息的唯⼀ID)，表明消息已经送达.
通过Publisher Confirms模式，⽣产者可以确保消息被RabbitMQ服务器成功接收, 从⽽避免消息丢失的问题.
适⽤场景: 对数据安全性要求较⾼的场景. ⽐如⾦融交易, 订单处理
⽣产者将信道设置成confirm(确认)模式, ⼀旦信道进⼊confirm模式, 所有在该信道上⾯发布的消息都会被指派⼀个唯⼀的ID(从1开始), ⼀旦消息被投递到所有匹配的队列之后, RabbitMQ就会发送⼀个确认给⽣产者(包含消息的唯⼀ID), 这就使得⽣产者知道消息已经正确到达⽬的队列了, 如果消息和队列是可持久化的, 那么确认消息会在将消息写⼊磁盘之后发出. broker回传给⽣产者的确认消息中deliveryTag 包含了确认消息的序号, 此外 broker 也可以设置channel.basicAck⽅法中的multiple参数, 表⽰到这个序号之前的所有消息都已经得到了处理.
使⽤发送确认机制, 必须要信道设置成confirm(确认)模式
发布确认有3种策略:
1. Publishing Messages Individually(单独确认)
public static void publishingMessagesIndividually() throws IOException, TimeoutException, InterruptedException {
try(Connection connection = createConnection()){
//创建channel
Channel channel = connection.createChannel();
//开启信道确认模式
channel.confirmSelect();
channel.queueDeclare(Common.PUBLISHER_CONFIRMS_QUEUE1, true, false, false, null);
Long start = System.currentTimeMillis();
for(int i = 0; i < MESSAGE_COUNT; i++){
String body = "消息" + i;
channel.basicPublish("", Common.PUBLISHER_CONFIRMS_QUEUE1, null, body.getBytes(StandardCharsets.UTF_8));
//等待确认消息, 只要消息被确认, 这个方法就会被返回
//如果超时过期, 则抛出TimeoutException, 如果任何消息被nack(丢失), waitForConfirmsOrDie将抛出IOException
channel.waitForConfirmsOrDie();
}
Long end = System.currentTimeMillis(5000);
System.out.printf("Published %d message individually in %d ms", MESSAGE_COUNT, end - start);
}
}
2. Publishing Messages in Batches(批量确认)
private static void publishingMessagesInBatches() throws IOException, TimeoutException, InterruptedException {
try(Connection connection = createConnection()){
Channel channel = connection.createChannel();
channel.confirmSelect();
channel.queueDeclare(Common.PUBLISHER_CONFIRMS_QUEUE2, true, false, false, null);
//批量个数
int batchSize = 100;
int outstandingMessageCount = 0;
long start = System.currentTimeMillis();
for(int i = 0; i < MESSAGE_COUNT; i++){
String body = "消息" + i;
channel.basicPublish("", Common.PUBLISHER_CONFIRMS_QUEUE2, null, body.getBytes(StandardCharsets.UTF_8));
outstandingMessageCount++;
if(outstandingMessageCount == batchSize){
channel.waitForConfirmsOrDie(5000);
outstandingMessageCount = 0;
}
}
if(outstandingMessageCount > 0){
channel.waitForConfirms(5000);
}
long end = System.currentTimeMillis();
System.out.printf("Published %d message batch in %d ms", MESSAGE_COUNT, end - start);
}
}
相⽐于单独确认策略, 批量确认极⼤地提升了confirm的效率, 缺点是出现Basic.Nack或者超时时, 我们不清楚具体哪条消息出了问题. 客⼾端需要将这⼀批次的消息全部重发, 这会带来明显的重复消息数量, 当消息经常丢失时,批量确认的性能应该是不升反降的
3. Handling Publisher Confirms Asynchronously(异步确认)
异步confirm⽅法的编程实现最为复杂. Channel 接⼝提供了⼀个⽅法addConfirmListener. 这个⽅法可以添加ConfirmListener 回调接⼝.
ConfirmListener 接⼝中包含两个⽅法: handleAck(long deliveryTag, boolean multiple) 和 handleNack(long deliveryTag, boolean multiple) , 分别对应处理RabbitMQ发送给⽣产者的ack和nack.
deliveryTag 表⽰发送消息的序号. multiple 表⽰是否批量确认.
我们需要为每⼀个Channel 维护⼀个已发送消息的序号集合. 当收到RabbitMQ的confirm 回调时, 从集合中删除对应的消息. 当Channel开启confirm模式后, channel上发送消息都会附带⼀个从1开始递增的deliveryTag序号. 我们可以使⽤SortedSet 的有序性来维护这个已发消息的集合.
- 当收到ack时, 从序列中删除该消息的序号. 如果为批量确认消息, 表⽰⼩于等于当前序号deliveryTag的消息都收到了, 则清除对应集合
- 当收到nack时, 处理逻辑类似, 不过需要结合具体的业务情况, 进⾏消息重发等操作
private static void handlingPublisherConfirmsAsynchronously() throws IOException, TimeoutException, InterruptedException{
try(Connection connection = createConnection()){
Channel channel = connection.createChannel();
channel.queueDeclare(Common.PUBLISHER_CONFIRMS_QUEUE3, true, false, false, null);
channel.confirmSelect();
//有序集合, 元素按照自然顺序进行排序, 存储未confirm消息序号
SortedSet<Long> confirmSet = Collections.synchronizedSortedSet(new TreeSet<>());
channel.addConfirmListener(new ConfirmListener() {
@Override
public void handleAck(long deliveryTag, boolean multiple) throws IOException {
if(multiple){
//批量
confirmSet.headSet(deliveryTag + 1).clear();
}else{
confirmSet.remove(deliveryTag);
}
}
@Override
public void handleNack(long deliveryTag, boolean multiple) throws IOException {
if(multiple){
//批量
confirmSet.headSet(deliveryTag + 1).clear();
}else{
confirmSet.remove(deliveryTag);
}
//如果处理失败, 需要有消息重发的环节, 此处省略
}
});
long start = System.currentTimeMillis();
for (int i = 0; i < MESSAGE_COUNT; i++) {
String message = "消息" + i;
long nextPublishSeqNo = channel.getNextPublishSeqNo();
channel.basicPublish("", Common.PUBLISHER_CONFIRMS_QUEUE3, null, message.getBytes(StandardCharsets.UTF_8));
confirmSet.add(nextPublishSeqNo);
}
while(!confirmSet.isEmpty()){
Thread.sleep(10);
}
long end = System.currentTimeMillis();
System.out.printf("Published %d message ConfirmsAsynchronously in %d ms", MESSAGE_COUNT, end - start);
}
}
