RabbitMQ 安裝延遲交換機delay-exchange

  • 2020 年 3 月 11 日
  • 筆記

安裝RabbitMQ 延遲交換機 在三台節點上安裝延遲交換機插件 1.進入RabbitMQ 插件目錄 

cd /usr/lib/rabbitmq/lib/rabbitmq_server-3.6.12/plugins

2.下載延遲插件安裝包 

wget https://dl.bintray.com/rabbitmq/community-plugins/3.6.x/rabbitmq_delayed_message_exchange/rabbitmq_delayed_message_exchange-20171215-3.6.x.zip

3.解壓延遲插件安裝包 

unzip rabbitmq_delayed_message_exchange--3.6.x.zip

將RAM 節點修改成DISK 節點 在 RAM 節點上，將節點修改成磁碟節點 

rabbitmqctl stop_app    rabbitmqctl change_cluster_node_type disk    rabbitmqctl start_app

啟動延遲交換機插件 在三台節點上啟動延遲交換機插件 

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

測試程式碼：

Producer:

import java.util.Date;  import java.util.HashMap;  import java.util.Map;    import com.rabbitmq.client.AMQP;  import com.rabbitmq.client.Channel;  import com.rabbitmq.client.Connection;  import com.rabbitmq.client.ConnectionFactory;      public class Producer {      public static void main(String[] args) throws Exception {          ConnectionFactory connectionFactory = new ConnectionFactory();          connectionFactory.setVirtualHost("/");          connectionFactory.setHost("172.31.1.135");          connectionFactory.setUsername("xx");          connectionFactory.setPassword("xx");          connectionFactory.setPort();          Connection connection = connectionFactory.newConnection();          Channel channel = connection.createChannel();          String exchangeName = "delay-exchange";          String routingkey = "delay.delay";            String queueName = "delay_queueName";          //x-delayed-message 聲明          Map<String,Object> map =new HashMap<>();          map.put("x-delayed-type", "direct");          channel.exchangeDeclare(exchangeName, "x-delayed-message", true, false, map);          //注意：arguments需要聲明在隊列上，聲明在交換機上是不會起作用的。          channel.queueDeclare(queueName, true, false, false, map);          channel.queueBind(queueName,exchangeName,routingkey);          for (int i = ; i < ; i++) {               // deliveryMode=2 持久化，expiration 消息有效時間              String msg = "delayed payload".getBytes("UTF-8") +" "+new Date().getTime();              byte[] messageBodyBytes = msg.getBytes();              Map<String, Object> headers = new HashMap<String, Object>();              headers.put("x-delay", );              AMQP.BasicProperties.Builder props = new AMQP.BasicProperties.Builder().headers(headers);              channel.basicPublish(exchangeName, routingkey, props.build(), messageBodyBytes);          }      }  }

Consumer:

import java.io.IOException;  import java.util.Date;    import com.rabbitmq.client.AMQP.BasicProperties;  import com.rabbitmq.client.Channel;  import com.rabbitmq.client.Connection;  import com.rabbitmq.client.ConnectionFactory;  import com.rabbitmq.client.DefaultConsumer;  import com.rabbitmq.client.Envelope;      public class Consumer {      public static void main(String[] args) throws Exception {          ConnectionFactory connectionFactory = new ConnectionFactory();          connectionFactory.setVirtualHost("/");          connectionFactory.setHost("172.31.1.135");          connectionFactory.setPort();          connectionFactory.setUsername("xxx");          connectionFactory.setPassword("xxx");          Connection connection = connectionFactory.newConnection();          Channel channel = connection.createChannel();          String queueName = "delay_queueName";          channel.queueDeclare(queueName,true,false,false,null);            channel.basicConsume(queueName, false, "myConsumer Tag", new DefaultConsumer(channel){              @Override              public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException {                    String routingKey = envelope.getRoutingKey();                    String convernType = properties.getContentType();                    long deliveryTag = envelope.getDeliveryTag();                    System.out.println("routingKey:"+routingKey+",convernType:"+convernType+",deliveryTag:"+deliveryTag+",Msg body:"+new String(body)+ " "+new Date().getTime());                    channel.basicAck(deliveryTag, false);              }            });      }  }

執行結果：

routingKey:delay.delay,convernType:null,deliveryTag:,Msg body:[B@c703b  routingKey:delay.delay,convernType:null,deliveryTag:,Msg body:[B@cbc42f  routingKey:delay.delay,convernType:null,deliveryTag:,Msg body:[B@c2f  routingKey:delay.delay,convernType:null,deliveryTag:,Msg body:[B@c703b  routingKey:delay.delay,convernType:null,deliveryTag:,Msg body:[B@cbc42f  routingKey:delay.delay,convernType:null,deliveryTag:,Msg body:[B@c2f

歡迎關註：【程式設計師開發者社區】

Tags:

VirMach 便宜 VPS