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

 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" );

     }

 }

RabbitMQ之消费者Demo(队列参数详细说明)的相关教程结束。