帮助类新增函数
public static AMQP.Queue.DeclareOk queueDeclare(Channel channel, String queueName, boolean durable) throws IOException {
return channel.queueDeclare(queueName, durable, false, false, null);
}
消费者
package com.dance.redis.mq.rabbit.qos;
import com.dance.redis.mq.rabbit.RabbitMQHelper;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeUnit;
public class Receiver {
public static void main(String[] args) throws Exception {
Channel channel = RabbitMQHelper.getChannel();
String queueName = "test001";
// durable 是否持久化消息
RabbitMQHelper.queueDeclare(channel,queueName,true);
channel.basicQos(0, 1, false);
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag,
Envelope envelope,
AMQP.BasicProperties properties,
byte[] body)
throws IOException {
System.out.println("receive message:" + new String(body) + ", RoutingKey: " + envelope.getRoutingKey());
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
channel.basicAck(envelope.getDeliveryTag(), false);
}
};
// 参数:队列名称、是否自动ACK、Consumer
channel.basicConsume(queueName, false, consumer);
// 等待回调函数执行完毕之后,关闭资源。
TimeUnit.SECONDS.sleep(50);
channel.close();
RabbitMQHelper.closeConnection();
}
}
生产者
package com.dance.redis.mq.rabbit.qos;
import com.dance.redis.mq.rabbit.RabbitMQHelper;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.util.HashMap;
import java.util.Map;
public class Sender {
public static void main(String[] args) throws Exception {
Channel channel = RabbitMQHelper.getChannel();
String queueName = "test001";
RabbitMQHelper.queueDeclare(channel, queueName, true);
Map<String, Object> headers = new HashMap<>();
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
.deliveryMode(2)
.contentEncoding("UTF-8")
.headers(headers).build();
for (int i = 0; i < 5; i++) {
String msg = "Hello World RabbitMQ " + i;
channel.basicPublish("", queueName, props, msg.getBytes());
}
}
}
启动消费者
启动生产者
查看消费者