RabbitMQ Routing
Routing模式：
package com.study.soufang.rabbit.a001.routing;
public class ConstantOfRouting {
public static final String QUEUE_NAME_A = "logs_routing_queue_a"; public static final String QUEUE_NAME_B = "logs_routing_queue_b"; public static final String EXCHANGE_NAME="logs_routing"; public static final String EXCHANGE_TYPE_FANOUT="fanout"; public static final String EXCHANGE_TYPE_ROUTING="direct"; }
package com.study.soufang.rabbit.a001.routing;
import java.io.IOException; import java.util.concurrent.TimeoutException;
import com.rabbitmq.client.Channel; import com.rabbitmq.client.DefaultConsumer; import com.rabbitmq.client.Envelope; import com.rabbitmq.client.AMQP.BasicProperties; import com.study.soufang.rabbit.a001.RabbitChannelUtil;
import lombok.Data; import lombok.experimental.Accessors; @Data @Accessors(chain = true) public class MyConsumer extends DefaultConsumer {
public MyConsumer(Channel channel) { super(channel); }
@Override public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException { try { String message = new String(body, "UTF-8"); System.out.println(consumerTag+"--"+Thread.currentThread().getName()+" [x] Received '" + message + "'"); } finally { /** * deliveryTag:該訊息的index multiple：是否批量.true:將一次性ack所有小於deliveryTag的訊息。 */ getChannel().basicAck(envelope.getDeliveryTag(), false); /*try { RabbitChannelUtil.closeChannel(getChannel()); } catch (TimeoutException e) { // TODO Auto-generated catch block e.printStackTrace(); }*/ } }
}
package com.study.soufang.rabbit.a001.routing;
import java.io.IOException; import java.util.concurrent.TimeoutException;
import com.rabbitmq.client.AMQP; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Consumer; import com.rabbitmq.client.DefaultConsumer; import com.rabbitmq.client.Envelope; import com.study.soufang.rabbit.a001.RabbitChannelUtil;
public class Recv {
public static void doRecv(String queue,String routingKey) throws IOException, TimeoutException{ Channel channel = null; try { channel = RabbitChannelUtil.createChannel(); //宣告交換機 channel.exchangeDeclare(ConstantOfRouting.EXCHANGE_NAME, ConstantOfRouting.EXCHANGE_TYPE_ROUTING); //宣告佇列 channel.queueDeclare(queue, false, false, false, null); //繫結佇列到交換機 channel.queueBind(queue, ConstantOfRouting.EXCHANGE_NAME, routingKey); Consumer consumer = new MyConsumer(channel); boolean autoAck = false; channel.basicConsume(queue, autoAck, consumer); channel.basicQos(1); } catch (IOException | TimeoutException e) { e.printStackTrace(); } } public static void main(String[] args) throws IOException, TimeoutException { new Thread(new Runnable() { @Override public void run() { try { doRecv(ConstantOfRouting.QUEUE_NAME_A,"error"); } catch (IOException e) { // TODO Auto-generated catch block e.printStackTrace(); } catch (TimeoutException e) { // TODO Auto-generated catch block e.printStackTrace(); } } }).start(); new Thread(new Runnable() { @Override public void run() { try { doRecv(ConstantOfRouting.QUEUE_NAME_B,"info"); } catch (IOException e) { // TODO Auto-generated catch block e.printStackTrace(); } catch (TimeoutException e) { // TODO Auto-generated catch block e.printStackTrace(); } } }).start(); new Thread(new Runnable() { @Override public void run() { try { doRecv(ConstantOfRouting.QUEUE_NAME_B,"debug"); } catch (IOException e) { // TODO Auto-generated catch block e.printStackTrace(); } catch (TimeoutException e) { // TODO Auto-generated catch block e.printStackTrace(); } } }).start(); } }
package com.study.soufang.rabbit.a001.routing;
import java.io.IOException; import java.util.concurrent.TimeoutException;
import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.study.soufang.rabbit.a001.RabbitChannelUtil;
public class Send {
public static void main(String[] args) throws IOException, TimeoutException { Channel channel = null; try { channel = RabbitChannelUtil.createChannel(); //宣告交換機 channel.exchangeDeclare(ConstantOfRouting.EXCHANGE_NAME, ConstantOfRouting.EXCHANGE_TYPE_ROUTING); //傳送訊息，注意：需要消費者先啟動並宣告佇列及繫結到此交換機，不然訊息會丟失 for(int i=0;i<100;i++){ String message = "log---"+i; String routingKey = null; if(i%3==0){ routingKey = "info"; }else if(i%5==0){ routingKey = "error"; }else{ routingKey = "debug"; } channel.basicPublish(ConstantOfRouting.EXCHANGE_NAME, routingKey, null, message.getBytes());; Thread.sleep(100); } } catch (IOException | TimeoutException | InterruptedException e) { e.printStackTrace(); }finally{ RabbitChannelUtil.closeChannel(channel); } } }
package com.study.soufang.rabbit.a001;
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 RabbitChannelUtil {
private final static String RABBIT_HOST= "192.168.10.22"; private final static int RABBIT_PORT=5672; public static Channel createChannel() throws IOException, TimeoutException{ ConnectionFactory factory = new ConnectionFactory(); factory.setHost(RABBIT_HOST); factory.setPort(RABBIT_PORT); factory.setUsername("admin"); factory.setPassword("admin"); factory.setVirtualHost("helloworld"); Connection connection = null; Channel channel = null; connection = factory.newConnection(); channel = connection.createChannel(); return channel; } /** * 關閉連線後不會再監聽 * @param channel * @throws IOException * @throws TimeoutException */ public static void closeChannel(Channel channel) throws IOException, TimeoutException{ Connection connection = channel.getConnection(); if(null != channel){ channel.close(); } if(null != connection){ connection.close(); } } }
