RabbitMQ四种Exchange类型之Direct (Java)

2019-07-04 08:35 分类：RabbitMQ 作者：云诺阅读(1313)

Direct类型的Exchange是不处理路由键，需要将一个队列绑定到交换机上，要求该消息与一个特定的路由键完全匹配。这是一个完整的匹配。如果一个队列绑定到该交换机上要求路由键为 “logs”，则只有路由键为“logs”的消息才被转发，不会转发路由键为"logs.error"，只会转发路由键为"logs"。

如下图：

QQ截图20190704200534.png

下面是代码实现！！！

消费者：

package direct;
 
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.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
 
public class DirectConsumer {
	private static final String EXCHANGE_NAME 	= "exchange_direct";
	public static void main(String[] argv) throws IOException, TimeoutException  {
		
		new ExchangeDirect("logs.info");
		new ExchangeDirect("logs.error");
		
	}
 
	static class ExchangeDirect{
		public  ExchangeDirect(String routingKey) throws IOException, TimeoutException {
			ConnectionFactory factory = new ConnectionFactory();
			//rabbitmq监听IP
			factory.setHost("192.168.249.128");
			//rabbitmq监听默认端口
			factory.setPort(5672);
			//设置访问的用户
			factory.setUsername("test");
			factory.setPassword("test");
			Connection connection = factory.newConnection();
			Channel channel = connection.createChannel();
			//声明路由名字和类型
			channel.exchangeDeclare(EXCHANGE_NAME, "direct", false, true, null);
			//队列名称
			String queueName = routingKey + ".queue";
			//创建队列
			channel.queueDeclare(queueName, false, false, true, null);
			//把队列绑定到路由上
			channel.queueBind(queueName, EXCHANGE_NAME, routingKey);
 
			System.out.println(" [routingKey = "+ routingKey +"] Waiting for msg....");
 
			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("[routingKey = "+ envelope.getRoutingKey() +"] Received msg is '" + message + "'");
				}
			};
			channel.basicConsume(queueName, true, consumer);
		}
 
	}
 
}

生产者：

package direct;
 
import java.io.IOException;
import java.util.concurrent.TimeoutException;
 
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
 
public class DirectProducer {
 
	private static final String EXCHANGE_NAME = "exchange_direct";
 
	public static void main(String[] argv) throws Exception{
		new ExchangeDirect("logs.info", "logs Info test ！！");
		new ExchangeDirect("logs.error", "logs error test ！！");
		new ExchangeDirect("logs.warning", "logs warning test ！！");
	}
	
	static class ExchangeDirect{
		public ExchangeDirect(String routingKey,String message) throws IOException, TimeoutException{
			ConnectionFactory factory = new ConnectionFactory();
			//rabbitmq监听IP
			factory.setHost("192.168.249.128");
			//rabbitmq监听默认端口
			factory.setPort(5672);
			//设置访问的用户
			factory.setUsername("test");
			factory.setPassword("test");
			Connection connection = factory.newConnection();
			Channel channel = connection.createChannel();
 
			//声明路由名字和类型
			channel.exchangeDeclare(EXCHANGE_NAME, "direct", false, true, null);
			
			channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes());
			System.out.println("[routingKey = "+routingKey+"] Sent msg is '" + message + "'");
 
			channel.close();
			connection.close();
			
		}
		
	}
 
}

运行结果：

QQ截图20190704203459.png