目錄

?

基本概念

代碼與實(shí)例

---

?

基本概念

生產(chǎn)者將信號(hào)設(shè)置成confirm模式，一旦信道進(jìn)入confirm模式，將有在該信道上面發(fā)布的消息都會(huì)被指派一個(gè)唯一的ID（從1開始），一旦消息被投遞到所有匹配的隊(duì)列后，RabbitMQ就會(huì)發(fā)送一個(gè)確認(rèn)給生產(chǎn)者（包含消息的唯一ID），這就使得生產(chǎn)者指導(dǎo)消息已經(jīng)正確到達(dá)目的隊(duì)列，如果消息和隊(duì)列是可持久化的，那么確認(rèn)消息將會(huì)消息寫入磁盤后，broker回傳給常指針的確認(rèn)消息中deliver-tag域包含了確認(rèn)消息的序列號(hào)。

confirm模式，開啟：

channel.confirmSelect();

confirm有3種模式：

1. 普通發(fā)一條：

waitForConfirms();

2. 批量發(fā)一堆：

waitForConfirms();

3. 異步confirm模式，提供了一個(gè)回調(diào)方法。

?

?

代碼與實(shí)例

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

單條發(fā)送：

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);//生產(chǎn)者調(diào)用 confirmSelect 將 channel 設(shè)置為 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();} }

批量發(fā)送：

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);//生產(chǎn)者調(diào)用 confirmSelect 將 channel 設(shè)置為 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

總結(jié)

以上是生活随笔為你收集整理的Java笔记-RabbitMQ中生产者端confirm模式的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。