RabbitMQ的萬用字元模式Demo
阿新 • • 發佈：2018-12-29
案例:
根據使用者的通知設定去通知使用者，設定接收Email的使用者只接收Email，設定接收sms的使用者只接收sms，設定兩種 通知型別都接收的則兩種通知都有效。
生產者：
package com.xuecheng.rabbitmq.producer;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory; 消費者：
package com.xuecheng.rabbitmq.consumer;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @Auther: 星仔
* @Date: 2018/12/24 22:43
* @Description:
*/
public class ConsumeTestSms {
//宣告佇列名稱
private static final String QUEUE_INFORM_SMS = "queue_inform_sms";
//宣告交換機的名稱
private static final String EXCHANGE_TOPICS_INFORM = "exchange_topics_inform";
//宣告RoutingKey
private static final String ROUTINGKEY_SMS = "inform.#.sms.#";
public static void main(String[] args) throws IOException, TimeoutException {
//建立連線工廠，建立連線
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setPort(5672);
factory.setUsername("guest");
factory.setPassword("guest");
//建立虛擬主機,rabbitmq預設虛擬機器名稱為“/”，虛擬機器相當於一個獨立的mq伺服器
factory.setVirtualHost("/");
//創建於MQ服務的TCP連線
Connection connection = factory.newConnection();
//創建於EXchange的通道，每一個通道都相當於一個會話事務
Channel channel = connection.createChannel();
/**
* 引數明細：
* exchange:交換機名稱
* type：交換機的型別
* fanout,topic,direct,headers
*/
channel.exchangeDeclare(EXCHANGE_TOPICS_INFORM, BuiltinExchangeType.TOPIC);
channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_TOPICS_INFORM,ROUTINGKEY_SMS);
//宣告佇列
channel.queueDeclare(QUEUE_INFORM_SMS,true,false,false,null);
//宣告消費訊息的方法
DefaultConsumer consumer = new DefaultConsumer(channel){
/**
* 消費者接收訊息呼叫此方法
* @param consumerTag 消費者的標籤，在channel.basicConsume()去指定
* @param envelope 訊息包的內容，可從中獲取訊息id，訊息routingkey，交換機，訊息和重傳標誌(收到訊息失敗後是否需要重新發送)
* @param properties 訊息屬性
* @param body 訊息內容
* @throws IOException
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
throws IOException{
//交換機
String exchange = envelope.getExchange();
//路由key
String routingKey = envelope.getRoutingKey();
//訊息id,mq在channel中用來標識訊息的id,用於確認訊息已接收
long deliveryTag = envelope.getDeliveryTag();
//訊息內容
String msg = new String(body,"utf-8"); System.out.println("receive message.." + msg);
}
};
/**
* 監聽佇列
* 引數：String queue, boolean autoAck,Consumer callback
* 引數明細：
* 1.queue： 佇列名稱
* 2.autoAck： 自動回覆，設定為true為表示訊息接收到自動向mq回覆接收到了，mq接收到回覆會刪除訊息，設定為false則需要手動回覆
* 3.callback： 消費訊息的方法，消費者接收到訊息後呼叫此方法
*/
channel.basicConsume(QUEUE_INFORM_SMS,true,consumer);
}
}
package com.xuecheng.rabbitmq.consumer;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @Auther: 星仔
* @Date: 2018/12/24 22:43
* @Description:
*/
public class ConsumeTestEmail {
//宣告佇列名稱
private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
//宣告交換機的名稱
private static final String EXCHANGE_TOPICS_INFORM = "exchange_topics_inform";
//宣告RoutingKey
private static final String ROUTINGKEY_EMAIL = "inform.#.email.#";
public static void main(String[] args) throws IOException, TimeoutException {
//建立連線工廠，建立連線
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setPort(5672);
factory.setUsername("guest");
factory.setPassword("guest");
//建立虛擬主機,rabbitmq預設虛擬機器名稱為“/”，虛擬機器相當於一個獨立的mq伺服器
factory.setVirtualHost("/");
//創建於MQ服務的TCP連線
Connection connection = factory.newConnection();
//創建於EXchange的通道，每一個通道都相當於一個會話事務
Channel channel = connection.createChannel();
/**
* 引數明細：
* exchange:交換機名稱
* type：交換機的型別
* fanout,topic,direct,headers
*/
channel.exchangeDeclare(EXCHANGE_TOPICS_INFORM, BuiltinExchangeType.TOPIC);
//宣告佇列
channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);
channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_TOPICS_INFORM,ROUTINGKEY_EMAIL);
//宣告消費訊息的方法
DefaultConsumer consumer = new DefaultConsumer(channel){
/**
* 消費者接收訊息呼叫此方法
* @param consumerTag 消費者的標籤，在channel.basicConsume()去指定
* @param envelope 訊息包的內容，可從中獲取訊息id，訊息routingkey，交換機，訊息和重傳標誌(收到訊息失敗後是否需要重新發送)
* @param properties 訊息屬性
* @param body 訊息內容
* @throws IOException
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
throws IOException{
//交換機
String exchange = envelope.getExchange();
//路由key
String routingKey = envelope.getRoutingKey();
//訊息id,mq在channel中用來標識訊息的id,用於確認訊息已接收
long deliveryTag = envelope.getDeliveryTag();
//訊息內容
String msg = new String(body,"utf-8"); System.out.println("receive message.." + msg);
}
};
/**
* 監聽佇列
* 引數：String queue, boolean autoAck,Consumer callback
* 引數明細：
* 1.queue： 佇列名稱
* 2.autoAck： 自動回覆，設定為true為表示訊息接收到自動向mq回覆接收到了，mq接收到回覆會刪除訊息，設定為false則需要手動回覆
* 3.callback： 消費訊息的方法，消費者接收到訊息後呼叫此方法
*/
channel.basicConsume(QUEUE_INFORM_EMAIL,true,consumer);
}
}