RabbitMQ之消费者Demo(队列参数详细说明)

时间:2021-03-09 06:00:38
 package com.jiefupay;

 import java.io.IOException;
import java.util.HashMap;
import java.util.Map; 8
import org.apache.log4j.Logger; import com.jiefupay.dao.Dao; import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope; public class App{ private static final Logger log = Logger.getLogger(App.class); 26
private static final String EXCHANGE_NAME = "refreshDispatcherMemoryExchange"; private static String QUEUE_NAME = "refreshDispatcherMemoryhfQueue"; public static void main(String[] args) throws Exception { ConnectionFactory factory = new ConnectionFactory();
factory.setHost("127.0.0.1");
factory.setPort(5672);
factory.setUsername("yourusername");
factory.setPassword("yourpassword"); //0.创建连接和通道
Connection connection = factory.newConnection();
Channel channel = connection.createChannel(); //1.声明一个死信交换机(扇形交换机)
channel.exchangeDeclare("refreshDispatcherDeadExchange", "fanout"); //2.创建队列的参数
Map<String, Object> queueArgs = new HashMap<String, Object>();
queueArgs.put("x-dead-letter-exchange", "refreshDispatcherDeadExchange"); //死信队列
queueArgs.put("x-message-ttl", 10000); // 消息超时:让发布的message在队列中可以存活多长时间,以毫秒为单位。
queueArgs.put("x-expires", 1000); // 队列超时:当前的queue在指定的时间内,没有消费者订阅就会被删除,以毫秒为单位。
queueArgs.put("x-max-length", 100); // 队列最大长度:当超过了这个大小的时候,会删除之前最早插入的消息为本次的留出空间。
queueArgs.put("x-queue-mode", "lazy"); //延迟加载:queue的信息尽可能的都保存在磁盘上,仅在有消费者订阅的时候才会加载到RAM中。 //3.声明队列。-将队列参数传到队列 (队列名字,是否持久化,是否排外,是否自动清理,参数)
channel.queueDeclare(QUEUE_NAME, true, false, false, queueArgs); //4.队列绑定交换机。 绑定键的意义依赖于转发器的类型,对于fanout类型,忽略此参数(第三个参数为binding key)。
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ""); Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
byte[] body) throws IOException {
// 捕获消息内容
String message = new String(body, "UTF-8"); try {
//消息处理(自己实现的方法)
messageHandler(message); //消息确认
channel.basicAck(envelope.getDeliveryTag(), false); }catch (Exception e) { //出现异常,置为true,重新入队。
channel.basicAck(envelope.getDeliveryTag(), true); //出现异常,不重新入队,而是reject入死信队列。
//channel.basicReject(envelope.getDeliveryTag(), false); }
}
};
//第二个参数值为false代表关闭RabbitMQ的自动应答机制,改为手动应答。
channel.basicConsume(QUEUE_NAME, false, consumer);
} public static void messageHandler(String message) {
switch (message) {
case "loadQDProductData": // 渠道信息 渠道产品
Dao.loadQDProductDataToSystem();
break;
case "loadQDGroupData": //渠道组
Dao.loadQDGroupDataToSystem();
break;
case "loadCustomerData": // 客户信息
Dao.loadCustomerDataToSystem();
break;
case "loadUserProductData": // 客户产品
Dao.loadUserProductDataToSystem();
break;
default:
break;
}
log.info( message + " Done" ); }
}