RabbitMQJava系列5-Routing模式

Posted 2020-12-25 zytiger

 

Routing模式只需要定义路由键即可

代码实现：

生产者

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.zy.rabbitmq.util.ConnectionUtil;

/**
* Routing模式
* 消息生产者
*
* @author zy
*
*/
public class Send {
private static final String EXCHAGNE_NAME = "test_exchange_direct";

public static void main(String[] args) throws IOException, TimeoutException {
// 获取连接
Connection connection = ConnectionUtil.getConnection();
// 从连接中获取一个通道
Channel channel = connection.createChannel();
// 交换机声明
channel.exchangeDeclare(EXCHAGNE_NAME, BuiltinExchangeType.DIRECT);
String msg = "Routing消息!";

// 绑定oragne路由键并发送
channel.basicPublish(EXCHAGNE_NAME, "oragne", null, msg.getBytes());

System.out.println("send Routing msg:" + msg);
channel.close();
connection.close();
}
}

 

消费者1

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.zy.rabbitmq.util.ConnectionUtil;

/**
* 消费者消费消息
*
* @author zy
*
*/
public class Receive {

private static final String QUEUE_NAME = "test_queue_direct1";

private static final String EXCHAGNE_NAME = "test_exchange_direct";

public static void main(String[] args) throws IOException, TimeoutException {
// 获取连接
Connection connection = ConnectionUtil.getConnection();
// 从连接中获取一个通道
final Channel channel = connection.createChannel();
// 队列声明
channel.queueDeclare(QUEUE_NAME, true, false, false, null);

// 绑定队列到交换机
channel.queueBind(QUEUE_NAME, EXCHAGNE_NAME, "oragne");

channel.basicQos(1);
// 基于事件
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");
System.out.println(" receive Routing oragne msg:" + message);
channel.basicAck(envelope.getDeliveryTag(), false);
}
};

// 监听队列
channel.basicConsume(QUEUE_NAME, false, consumer);
}

}

 

消费者2

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.zy.rabbitmq.util.ConnectionUtil;

/**
* 消费者消费消息
*
* @author zy
*
*/
public class Receive2 {

private static final String QUEUE_NAME = "test_queue_direct2";

private static final String EXCHAGNE_NAME = "test_exchange_direct";

public static void main(String[] args) throws IOException, TimeoutException {
// 获取连接
Connection connection = ConnectionUtil.getConnection();
// 从连接中获取一个通道
final Channel channel = connection.createChannel();
// 队列声明
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
// 绑定队列到交换机，并绑定error路由键
channel.queueBind(QUEUE_NAME, EXCHAGNE_NAME, "black");

channel.queueBind(QUEUE_NAME, EXCHAGNE_NAME, "green");

channel.basicQos(1);
// 基于事件
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");
System.out.println(" receive Routing black or green msg:" + message);
channel.basicAck(envelope.getDeliveryTag(), false);
}
};

// 监听队列
channel.basicConsume(QUEUE_NAME, false, consumer);
}

}

生产者指定了路由键为orange

消费者1指定orange的路由键，因此能接收到消息

消费者2指定black和green的路由键，因此不能接收到消息

参考：http://www.rabbitmq.com/tutorials/tutorial-four-java.html