RabbitMQ-主题模式Topic

将路由和某个模式匹配
# 匹配一个或者多个
* 匹配一个

例如 Good.insert Good.delete
Good.#都能获得

商品 : 发布 删除 修改 查询....

package com.ithzk.rabbitmq.topic;

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_topic";

public static void main(String[] args) throws IOException, TimeoutException {

Connection connection = RabbitMQConnectionUtils.getConnection();

Channel channel = connection.createChannel();

// 声明交换机
channel.exchangeDeclare(EXCHANGER_NAME,"topic");

String msg = "goods......";

//发送goods.add 消费者1和消费者2都可以收到消息
//发送goods.delete 消费者2收到消息
String routingKey = "goods.delete";
channel.basicPublish(EXCHANGER_NAME,routingKey,null,msg.getBytes());

System.out.println("Send msg"+msg);

channel.close();
connection.close();
}

}

package com.ithzk.rabbitmq.topic;

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_topic_one";
private final static String EXCHANGER_NAME = "test_exchange_topic";

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 = "goods.add";
//绑定队列到交换机 转发器
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]");

}

}

package com.ithzk.rabbitmq.topic;

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_topic_two";
private final static String EXCHANGER_NAME = "test_exchange_topic";

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 = "goods.#";
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("[2] Recv msg:" + msg);

try {
Thread.sleep(10000);
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
System.out.println("[2] done"<
c635
/span>);
channel.basicAck(envelope.getDeliveryTag(), false);
}
}
};

boolean autoAck = false;

channel.basicConsume(QUEUE_NAME,autoAck,consumer);

System.out.println("[Consumer 2 start]");

}

}