当前位置：首页 > news >正文

java如何实现rabbitmq的消息确认机制和消息持久化机制配置和示例

news 2025/7/21 1:30:06

在Java中，使用RabbitMQ的客户端库（通常是AMQP客户端库，如RabbitMQ的Java客户端）可以方便地实现消息确认机制和消息持久化机制。以下是如何实现这两个机制的示例。

1、消息确认机制

RabbitMQ支持两种类型的确认：生产者到交换机的确认（通常在发送时默认进行）和消费者到队列的确认。在Java中，消费者确认通常通过手动确认消息来实现。

1、生产者示例

import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Channel;public class Producer {private final static String QUEUE_NAME = "my_queue";public static void main(String[] args) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");try (Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {channel.queueDeclare(QUEUE_NAME, false, false, false, null);String message = "Hello World!";channel.basicPublish("", QUEUE_NAME, null, message.getBytes("UTF-8"));System.out.println(" [x] Sent '" + message + "'");}}
}

2、消费者示例（使用手动确认）

import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer {private final static String QUEUE_NAME = "my_queue";public static void main(String[] argv) throws IOException, TimeoutException {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 + "'");// 手动确认消息try {channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);} catch (IOException e) {e.printStackTrace();}};channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> { });}
}

在上面的消费者示例中，basicConsume方法的第二个参数设置为false，表示消息不会自动确认。当消息被处理完成后，调用basicAck方法来手动确认消息。

2、消息持久化机制

要使消息持久化，你需要确保队列、消息以及交换机都是持久化的。

1、持久化队列和消息示例

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class DurableProducer {private final static String QUEUE_NAME = "my_durable_queue";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(QUEUE_NAME, true, false, false, null);String message = "Hello World!";// 设置消息的属性为持久化BasicProperties properties = new BasicProperties.Builder().deliveryMode(2) // 设置为2表示消息是持久化的.build();channel.basicPublish("", QUEUE_NAME, properties, message.getBytes("UTF-8"));System.out.println(" [x] Sent '" + message + "'");}}
}