1.Exchange
交換機，轉發器
一方面接收生產者的消息，另一方面向隊列推送消息
匿名轉發(第一個參數為"")
channel.basicPublish("","",null,msg.getBytes());
fanout(不處理路由鍵)
每個和交換機綁定的隊列都會收到消息
channel.exchangeDeclare(EXCHANGER_NAME,"fanout");
channel.basicPublish(EXCHANGER_NAME,"",null,msg.getBytes());
direct(處理路由鍵)
會根據發送的key鍵發送給對應隊列
路由模式
例如發送消息帶有error key，則兩個隊列都會收到消息
若只發送info key，下方隊列會收到消息
生產者
package com.ithzk.rabbitmq.routing;
import com.ithzk.rabbitmq.utils.RabbitMQConnectionUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @author hzk
* @date 2018/3/10
*/
public class Send {
private final static String EXCHANGER_NAME = "test_exchange_direct";
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = RabbitMQConnectionUtils.getConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare(EXCHANGER_NAME,"direct");
String msg = "hello exchange direct";
//發送error兩個隊列都會收到消息
String routingKey = "error";
//發送info 只有消費者2會收到消息
String routingKey2 = "info";
channel.basicPublish(EXCHANGER_NAME,routingKey2,null,msg.getBytes());
System.out.println("Send msg"+msg);
channel.close();
connection.close();
}
}
消費者1
package com.ithzk.rabbitmq.routing;
import com.ithzk.rabbitmq.utils.RabbitMQConnectionUtils;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @author hzk
* @date 2018/3/10
*/
public class Recv1 {
private static final String QUEUE_NAME="test_queue_direct_one";
private final static String EXCHANGER_NAME = "test_exchange_direct";
public static void main(String[] args) throws IOException, TimeoutException {
//獲取連接
Connection connection = RabbitMQConnectionUtils.getConnection();
//從連接中獲取頻道
final Channel channel = connection.createChannel();
channel.queueDeclare(QUEUE_NAME,false,false,false,null);
String routingKey = "error";
//綁定隊列到交換機 轉發器
channel.queueBind(QUEUE_NAME,EXCHANGER_NAME,routingKey);
//保證一次只發一個
channel.basicQos(1);
DefaultConsumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String msg = new String(body, "utf-8");
System.out.println("[1] Recv msg:" + msg);
try {
Thread.sleep(10000);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
System.out.println("[1] done");
channel.basicAck(envelope.getDeliveryTag(), false);
}
}
};
boolean autoAck = false;
channel.basicConsume(QUEUE_NAME,autoAck,consumer);
System.out.println("[Consumer 1 start]");
}
}
消費者2
package com.ithzk.rabbitmq.routing;
import com.ithzk.rabbitmq.utils.RabbitMQConnectionUtils;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @author hzk
* @date 2018/3/10
*/
public class Recv2 {
private static final String QUEUE_NAME="test_queue_direct_two";
private final static String EXCHANGER_NAME = "test_exchange_direct";
public static void main(String[] args) throws IOException, TimeoutException {
//獲取連接
Connection connection = RabbitMQConnectionUtils.getConnection();
//從連接中獲取頻道
final Channel channel = connection.createChannel();
channel.queueDeclare(QUEUE_NAME,false,false,false,null);
//綁定隊列到交換機 轉發器
String routingKey = "error";
String routingKey2 = "info";
String routingKey3 = "warning";
channel.queueBind(QUEUE_NAME,EXCHANGER_NAME,routingKey);
channel.queueBind(QUEUE_NAME,EXCHANGER_NAME,routingKey2);
channel.queueBind(QUEUE_NAME,EXCHANGER_NAME,routingKey3);
//保證一次只發一個
channel.basicQos(1);
DefaultConsumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String msg = new String(body, "utf-8");
System.out.println("[2] Recv msg:" + msg);
try {
Thread.sleep(10000);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
System.out.println("[2] done");
channel.basicAck(envelope.getDeliveryTag(), false);
}
}
};
boolean autoAck = false;
channel.basicConsume(QUEUE_NAME,autoAck,consumer);
System.out.println("[Consumer 2 start]");
}
}
閱讀更多 丶灣裡小白龍 的文章
關鍵字: 技術
