中间件技术JAVA原生队列
·
安装延迟队列插件
先看 RabbitMQ 版本
插件版本必须和 RabbitMQ 主版本一致（例如 MQ 是 3.13.x，插件就下 3.13.x）。
下载地址（选对应版本的 .ez 文件）：
https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases
找到插件目录
默认类似：
C:\Program Files\RabbitMQ Server\rabbitmq_server-3.13.0\plugins
（版本号按你实际的来）
把下载好的 .ez 文件直接丢进 plugins 目录，不要解压。
打开命令行，进入 sbin 目录：
cd /d D:\zhongjianjian\rabbitMQ\rabbitmq_server-4.2.4\sbin
启用插件：
rabbitmq-plugins enable rabbitmq_delayed_message_exchange
重启 RabbitMQ 服务：
rabbitmq-service stop
rabbitmq-service start
原生客户端依赖：
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.25.0</version>
</dependency>Simple队列生产者
public class Producer {
public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setPort(5672);
factory.setVirtualHost("/");
factory.setUsername("guest");
factory.setPassword("guest");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.queueDeclare("test.simple_queue", true, false, false, null);
String massage = "hello rabbitmq";
channel.basicPublish("", "test.simple_queue", null, massage.getBytes());
channel.close();
connection.close();
}
}
simple消费者：
public class Comsumer {
public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setPort(5672);
factory.setVirtualHost("/");
factory.setUsername("guest");
factory.setPassword("guest");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.queueDeclare("test.simple_queue", true, false, false, null);
DefaultConsumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String comsumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("consumerTag: " + comsumerTag);
System.out.println("Exchange: " + envelope.getExchange());
System.out.println("RoutingKey: " + envelope.getRoutingKey());
System.out.println("properties: " + properties);
System.out.println("body: " + new String(body));
}
};
channel.basicConsume("test.simple_queue", true, consumer);
}
}Work生产者：
public class WorkProducer {
private static final String QName = "test.work.queue";
public static void main(String[] args) throws Exception{
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setPort(5672);
factory.setVirtualHost("/");
factory.setUsername("guest");
factory.setPassword("guest");
try(Connection connection = factory.newConnection();
Channel channel = connection.createChannel()){
channel.queueDeclare(QName,true,false,false,null);
for(int i=0;i<10;i++){
String msg = "Task " + i + " (耗时" + (i % 3 + 1) + "秒)";
channel.basicPublish("", QName, null,msg.getBytes());
System.out.println("[生产者] 发送任务: " + msg);
Thread.sleep(500);
}
}
}
}
Work消费者：
public class WorkComsumer {
private static final String QName = "test.work.queue";
public static void main(String[] args) throws Exception{
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setPort(5672);
factory.setVirtualHost("/");
factory.setUsername("guest");
factory.setPassword("guest");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.queueDeclare(QName,true,false,false,null);
channel.basicQos(1);
DefaultConsumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String comsumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body)throws IOException {
String message = new String(body, StandardCharsets.UTF_8);
System.out.println("[消费者] 处理任务: " + message);
try {
Thread.sleep(1000 * Integer.parseInt(message.split("耗时")[1].split("秒")[0]));
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
channel.basicAck(envelope.getDeliveryTag(),false);
System.out.println("[消费者] 任务完成: " + message);
}
};
channel.basicConsume(QName,false,consumer);
}
}
Topic生产者：
public class TopicProducer {
private static final String ExchangeName = "test.topic.logs";
public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setPort(5672);
factory.setVirtualHost("/");
factory.setUsername("guest");
factory.setPassword("guest");
try(Connection connection = factory.newConnection();
Channel channel = connection.createChannel()){
channel.exchangeDeclare(ExchangeName, "topic");
String[] routingKeys = {"stock.usd.eur", "stock.eur.usd", "stock.usd", "stock.eur"};
for(String routingkey : routingKeys){
String message = "task" + routingkey;
channel.basicPublish(ExchangeName, routingkey, null, message.getBytes());
System.out.println("[生产者] 发送消息: " + message);
}
}
}
}
Topic消费者：
public class TopicComsumer {
private static final String ExchangeName = "test.topic.logs";
public static void main(String[] args) throws Exception{
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setPort(5672);
factory.setVirtualHost("/");
factory.setUsername("guest");
factory.setPassword("guest");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
String qu = channel.queueDeclare().getQueue();
channel.exchangeDeclare(ExchangeName, "topic");
String[] bindkey = {"stock.*", "stock.#", "stock.usd.*", "stock.*.eur"};
for(String key : bindkey){
channel.queueBind(qu,ExchangeName,key);
System.out.println("绑定队列: " + qu + " 到绑定键: " + key);
}
System.out.println("等待接收通配符消息...");
DefaultConsumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String comsumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body){
String message = new String(body, StandardCharsets.UTF_8);
System.out.println("[消费者] 收到消息: " + message);
}
};
channel.basicConsume(qu, true, consumer);
}
}
Routing生产者：
public class RoutingProducer {
public static final String EN = "test.routing.exchange";
public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setPort(5672);
factory.setVirtualHost("/");
factory.setUsername("guest");
factory.setPassword("guest");
try(Connection connection = factory.newConnection();
Channel channel = connection.createChannel()){
channel.exchangeDeclare(EN, "direct");
String[] routingKeys = {"info", "warning", "error"};
for(String routingKey : routingKeys){
String mag = "Log message with routing key: " + routingKey;
channel.basicPublish(EN, routingKey, null, mag.getBytes());
System.out.println("[生产者] 发送消息: " + mag);
}
}
}
}
Routing消费者：
public class RotingComsumer {
private static final String EXCHANGE_NAME = "test.routing.exchange";
public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setPort(5672);
factory.setVirtualHost("/");
factory.setUsername("guest");
factory.setPassword("guest");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME, "direct");
String qu = channel.queueDeclare().getQueue();
String[] routingKeys = {"info", "warning", "error"};
for (String routingKey : routingKeys) {
channel.queueBind(qu, EXCHANGE_NAME, routingKey);
System.out.println("绑定队列: " + qu + " 到路由键: " + routingKey);
}
System.out.println("等待接收路由消息...");
DefaultConsumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String comsumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, StandardCharsets.UTF_8);
System.out.println("[消费者] 收到消息: " + message);
}
};
channel.basicConsume(qu, true, consumer);
}
}
Publish生产者：
public class PublishProducer {
public static final String EN = "test.publish";
public static void main(String[] args) throws Exception{
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setPort(5672);
factory.setVirtualHost("/");
factory.setUsername("guest");
factory.setPassword("guest");
try(Connection connection = factory.newConnection();
Channel channel = connection.createChannel()){
channel.exchangeDeclare(EN, "fanout");
for (int i = 1; i <= 5; i++) {
String message = "Message " + i;
channel.basicPublish(EN, "", null, message.getBytes());
System.out.println("[生产者] 发送消息: " + message);
Thread.sleep(1000);
}
}
}
}
Publish消费者：
public class PublishComsumer {
public static final String EN = "test.publish";
public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setPort(5672);
factory.setVirtualHost("/");
factory.setUsername("guest");
factory.setPassword("guest");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EN, "fanout");
String qn = channel.queueDeclare("", true, false, true, null).getQueue();
channel.queueBind(qn, EN, "");
System.out.println("等待接收发布订阅消息...");
DefaultConsumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery (String comsumerTag, Envelope envelope,
AMQP.BasicProperties properties, byte[] body){
String message = new String(body, StandardCharsets.UTF_8);
System.out.println("[消费者] 收到消息: " + message);
}
};
channel.basicConsume(qn, true, consumer);
}
}
Delay生产者：
public class DelayQueueProducer {
// RabbitMQ 连接信息
private static final String HOST = "127.0.0.1";
private static final int PORT = 5672;
private static final String USERNAME = "guest";
private static final String PASSWORD = "guest";
// 交换机、队列、路由键（和消费者完全一致）
private static final String EXCHANGE_NAME = "exchange.delay.video";
private static final String QUEUE_NAME = "queue.delay.video";
private static final String ROUTING_KEY = "routing.key.delay.video";
public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost(HOST);
factory.setPort(PORT);
factory.setUsername(USERNAME);
factory.setPassword(PASSWORD);
try (Connection connection = factory.newConnection("delay-producer");
Channel channel = connection.createChannel()) {
Map<String, Object> delayArgs = new HashMap<>();
// x-delayed-type 指定底层真实的路由规则（这里配合你的 ROUTING_KEY 使用 direct）
delayArgs.put("x-delayed-type", "direct");
// 2. 声明类型为 "x-delayed-message" 的特殊交换机
// 注意：第二个参数必须是字符串 "x-delayed-message"，不能是 BuiltinExchangeType
channel.exchangeDeclare(EXCHANGE_NAME, "x-delayed-message", true, false, delayArgs);
// 3. 声明并绑定普通的持久化队列（保证环境就绪）
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY)
// 延迟消息：设置消息头 x-delay（单位：毫秒）
Map<String, Object> headers = new HashMap<>();
headers.put("x-delay", 10000); // 延迟10秒
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
.contentType("text/plain")
.deliveryMode(2) // 持久化消息
.headers(headers)
.build();
String now = new SimpleDateFormat("HH:mm:ss").format(new Date());
String message = "测试基于插件的延迟消息 [" + now + "]";
channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, props, message.getBytes(StandardCharsets.UTF_8));
System.out.println("[生产者] 发送时间: " + now);
System.out.println("[生产者] 消息内容: " + message);
System.out.println("[生产者] 延迟: 10秒");
}
}
}
Delay 消费者：
public class DelayQueueConsumer {
// RabbitMQ 连接信息
private static final String HOST = "127.0.0.1";
private static final int PORT = 5672;
private static final String USERNAME = "guest";
private static final String PASSWORD = "guest";
// 交换机、队列、路由键
private static final String EXCHANGE_NAME = "exchange.delay.video";
private static final String QUEUE_NAME = "queue.delay.video";
private static final String ROUTING_KEY = "routing.key.delay.video";
public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost(HOST);
factory.setPort(PORT);
factory.setUsername(USERNAME);
factory.setPassword(PASSWORD);
Connection connection = factory.newConnection("delay-consumer");
Channel channel = connection.createChannel();
// 声明延迟交换机、队列、绑定
declareDelayEnvironment(channel);
// 启动消费者
startConsumer(channel);
System.out.println("========== 延迟消费者已启动，等待消息 ==========");
}
/**
* 声明延迟交换机、队列、绑定
*/
private static void declareDelayEnvironment(Channel channel) throws Exception {
// 延迟交换机参数
Map<String, Object> exchangeArgs = new HashMap<>();
exchangeArgs.put("x-delayed-type", "direct");
// 声明延迟交换机
channel.exchangeDeclare(
EXCHANGE_NAME,
"x-delayed-message",
true,
false,
exchangeArgs
);
// 普通队列
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
// 绑定
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
}
/**
* 启动消费者（手动ACK）
*/
private static void startConsumer(Channel channel) throws Exception {
channel.basicQos(1); // 每次消费1条
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
long deliveryTag = delivery.getEnvelope().getDeliveryTag();
String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
String now = new SimpleDateFormat("HH:mm:ss").format(new Date());
try {
System.out.println("\n[消费者] 接收时间: " + now);
System.out.println("[消费者] 收到消息: " + message);
channel.basicAck(deliveryTag, false);
System.out.println("[消费者] ACK 成功：" + deliveryTag);
} catch (Exception e) {
channel.basicNack(deliveryTag, false, true);
System.out.println("[消费者] 消费失败，重回队列");
}
};
CancelCallback cancelCallback = consumerTag ->
System.out.println("消费者被取消：" + consumerTag);
channel.basicConsume(QUEUE_NAME, false, deliverCallback, cancelCallback);
}
}
Lazy生产者：
public class LazyQueueProducer {
private static final String HOST = "127.0.0.1";
private static final int PORT = 5672;
private static final String USERNAME = "guest";
private static final String PASSWORD = "guest";
private static final String EXCHANGE_NAME = "exchange.lazy.test";
private static final String QUEUE_NAME = "queue.lazy.test";
private static final String ROUTING_KEY = "routing.key.lazy.test";
public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost(HOST);
factory.setPort(PORT);
factory.setUsername(USERNAME);
factory.setPassword(PASSWORD);
Connection connection = factory.newConnection("lazy-producer");
Channel channel = connection.createChannel();
// 声明队列（保证存在）
Map<String, Object> queueArgs = new HashMap<>();
queueArgs.put("x-queue-mode", "lazy");
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT, true);
channel.queueDeclare(QUEUE_NAME, true, false, false, queueArgs);
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
// 发送5条消息
for (int i = 1; i <= 5; i++) {
String msg = "惰性队列测试消息_" + i;
channel.basicPublish(
EXCHANGE_NAME,
ROUTING_KEY,
MessageProperties.PERSISTENT_TEXT_PLAIN,
msg.getBytes(StandardCharsets.UTF_8)
);
System.out.println("生产者发送：" + msg);
Thread.sleep(1000);
}
System.out.println("========== 消息发送完成 ==========");
channel.close();
connection.close();
}
}Lazy消费者：
public class LazyQueueConsumer {
// RabbitMQ 连接信息
private static final String HOST = "127.0.0.1";
private static final int PORT = 5672;
private static final String USERNAME = "guest";
private static final String PASSWORD = "guest";
// 交换机、队列、路由键（必须与生产者一致）
private static final String EXCHANGE_NAME = "exchange.lazy.test";
private static final String QUEUE_NAME = "queue.lazy.test";
private static final String ROUTING_KEY = "routing.key.lazy.test";
public static void main(String[] args) throws Exception {
// 1. 创建连接工厂
ConnectionFactory factory = new ConnectionFactory();
factory.setHost(HOST);
factory.setPort(PORT);
factory.setUsername(USERNAME);
factory.setPassword(PASSWORD);
// 2. 创建连接和通道
Connection connection = factory.newConnection("lazy-queue-consumer");
Channel channel = connection.createChannel();
// 3. 声明交换机、惰性队列、绑定
declareLazyQueue(channel);
// 4. 启动消费（手动确认）
startConsumer(channel);
System.out.println("========== 惰性队列消费者已启动，等待消息 ==========");
}
/**
* 声明惰性队列（x-queue-mode=lazy）
*/
private static void declareLazyQueue(Channel channel) throws Exception {
// 声明交换机
channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT, true);
// 惰性队列参数
Map<String, Object> args = new HashMap<>();
args.put("x-queue-mode", "lazy");
// 声明队列
channel.queueDeclare(QUEUE_NAME, true, false, false, args);
// 绑定
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
}
/**
* 启动消费者：手动确认
*/
private static void startConsumer(Channel channel) throws Exception {
// 每次消费1条
channel.basicQos(1);
// 消息接收回调
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), StandardCharsets.UTF_8);
long deliveryTag = delivery.getEnvelope().getDeliveryTag();
try {
System.out.println("消费者收到：" + message);
// 模拟业务处理
Thread.sleep(500);
// 手动确认
channel.basicAck(deliveryTag, false);
System.out.println("消息已确认，deliveryTag=" + deliveryTag);
} catch (Exception e) {
// 消费失败，重回队列
channel.basicNack(deliveryTag, false, true);
System.out.println("消费失败，消息重回队列：" + message);
}
};
// 取消回调
CancelCallback cancelCallback = consumerTag ->
System.out.println("消费者被取消：" + consumerTag);
// 关闭自动确认
channel.basicConsume(QUEUE_NAME, false, deliverCallback, cancelCallback);
}
}更多推荐
5
0- 0
已为社区贡献2条内容
所有评论(0)