### 消費者1 ``` 我們此處假設消費者1只接收兩種類型的消息：更新商品和刪除商品 ``` ~~~ public static void main(String[] argv) throws Exception { // 獲取到連接 Connection connection = ConnectionUtil.getConn(); // 獲取通道 Channel channel = connection.createChannel(); // 聲明隊列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 綁定隊列到交換機，同時指定需要訂閱的routing key。假設此處需要update和delete消息 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update"); channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete"); // 定義隊列的消費者 DefaultConsumer consumer = new DefaultConsumer(channel) { // 獲取消息，并且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用 @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { // body 即消息體 String msg = new String(body); System.out.println(" [消費者1] received : " + msg + "!"); } }; // 監聽隊列，自動ACK channel.basicConsume(QUEUE_NAME, true, consumer); } } ~~~ ### 消費者2 ``` 我們此處假設消費者2接收所有類型的消息：新增商品，更新商品和刪除商品。 ``` ~~~ public class Recv2 { private final static String QUEUE_NAME = "direct_exchange_queue_2"; private final static String EXCHANGE_NAME = "direct_exchange_test"; public static void main(String[] argv) throws Exception { // 獲取到連接 Connection connection = ConnectionUtil.getConnection(); // 獲取通道 Channel channel = connection.createChannel(); // 聲明隊列 channel.queueDeclare(QUEUE_NAME, false, false, false, null); // 綁定隊列到交換機，同時指定需要訂閱的routing key。訂閱 insert、update、delete channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "insert"); channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "update"); channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "delete"); // 定義隊列的消費者 DefaultConsumer consumer = new DefaultConsumer(channel) { // 獲取消息，并且處理，這個方法類似事件監聽，如果有消息的時候，會被自動調用 @Override public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException { // body 即消息體 String msg = new String(body); System.out.println(" [消費者2] received : " + msg + "!"); } }; // 監聽隊列，自動ACK channel.basicConsume(QUEUE_NAME, true, consumer); } } ~~~