目录

 

基本概念

代码与实例

---

 

基本概念

生产者将信号设置成confirm模式，一旦信道进入confirm模式，将有在该信道上面发布的消息都会被指派一个唯一的ID（从1开始），一旦消息被投递到所有匹配的队列后，RabbitMQ就会发送一个确认给生产者（包含消息的唯一ID），这就使得生产者指导消息已经正确到达目的队列，如果消息和队列是可持久化的，那么确认消息将会消息写入磁盘后，broker回传给常指针的确认消息中deliver-tag域包含了确认消息的序列号。

confirm模式，开启：

channel.confirmSelect();

confirm有3种模式：

1. 普通发一条：

waitForConfirms();

2. 批量发一堆：

waitForConfirms();

3. 异步confirm模式，提供了一个回调方法。

 

 

代码与实例

Recv.java

package confirm;import com.rabbitmq.client.*; import util.ConnectionUtils;import java.io.IOException; import java.util.concurrent.TimeoutException;public class Recv {private static final String QUEUE_NAME = "test_queue_confirm1";public static void main(String[] args) throws IOException, TimeoutException {Connection connection = ConnectionUtils.getConnect();Channel channel = connection.createChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);channel.basicConsume(QUEUE_NAME, true, new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("recv[confirm] msg : " + new String(body, "utf-8"));}});} }

ConnectionUtils.java

package util;import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory;import java.io.IOException; import java.util.concurrent.TimeoutException;public class ConnectionUtils {public static Connection getConnect() throws IOException, TimeoutException {ConnectionFactory factory = new ConnectionFactory();factory.setHost("127.0.0.1");factory.setPort(5672);factory.setVirtualHost("/vhost_cff");factory.setUsername("cff");factory.setPassword("123");return factory.newConnection();} }

单条发送：

send1.java

package confirm;import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import util.ConnectionUtils;import java.io.IOException; import java.util.concurrent.TimeoutException;public class Send1 {private static final String QUEUE_NAME = "test_queue_confirm1";public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {Connection connection = ConnectionUtils.getConnect();Channel channel = connection.createChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);//生产者调用 confirmSelect 将 channel 设置为 confirm 模式注意channel.confirmSelect();String msgString = "hello confirm message !";channel.basicPublish("", QUEUE_NAME, null, msgString.getBytes());if(!channel.waitForConfirms()){System.out.println("message send failed !");}else{System.out.println("message send ok !");}channel.close();connection.close();} }

批量发送：

send2.java

package confirm;import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import util.ConnectionUtils;import java.io.IOException; import java.util.concurrent.TimeoutException;public class Send2 {private static final String QUEUE_NAME = "test_queue_confirm1";public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {Connection connection = ConnectionUtils.getConnect();Channel channel = connection.createChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);//生产者调用 confirmSelect 将 channel 设置为 confirm 模式注意channel.confirmSelect();String msgString = "hello confirm message !";for(int i = 0; i < 100; i++){channel.basicPublish("", QUEUE_NAME, null, msgString.getBytes());}if(!channel.waitForConfirms()){System.out.println("message send failed !");}else{System.out.println("message send ok !");}channel.close();connection.close();} }

下载地址：

https://github.com/fengfanchen/Java/tree/master/RabbitMQComfirm

总结

以上是生活随笔为你收集整理的Java笔记-RabbitMQ中生产者端confirm模式的全部内容，希望文章能够帮你解决所遇到的问题。

如果觉得生活随笔网站内容还不错，欢迎将生活随笔推荐给好友。