目錄

?

基本概念

代碼與實例

---

?

基本概念

生產者將信號設置成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模式的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。