开发笔记:发布订阅模式

篇首语：本文由编程笔记#小编为大家整理，主要介绍了发布订阅模式相关的知识，希望对你有一定的参考价值。

• Publish/Subscribe(发布订阅模式)

  发布订阅模式其实就是生产者将数据发送到交换机，交换机将所有的消息发送到每个绑定的队列中，因此 在发布消息时可以只先指定交换机的名称，交换机的声明的代码可以放到消费者端进行声明，队列的声明也放在消费者端来声明

  
  
  
  

  • Exchange类型-fanout fanout类型的Exchange路由规则非常简单，它会把所有发送到该Exchange的消息路由到所有与它绑定的Queue中

    

    在使用发布订阅模式的时候，我们只需要声明该队列为fanout即可。如下：

    channel.exchangeDeclare("logs", "fanout");

  
  

  • Temporary Queues(临时队列)

  
  

  • 下面代码是两个消费者和一个生产者实现发布订阅模式

    生产者代码:

    public class PSEmitLog {
private static final String EXCHANGE_NAME = "logs";
public static void main(String[] args) throws Exception {
//获取连接
Connection cOnnection= ConnectionUtil.getConnection("localhost", 5672, "/", "guest", "guest");
Channel channel = connection.createChannel();
//声明交换机，发布订阅模式
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
//发送消息
for (int i = 0; i <10; i++) {
String message = " message" + i;
System.out.println("[send]：" + message);
//发送消息
channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("utf-8"));
}
channel.close();
connection.close();
}
}

    消费者代码：

    public class PSEmitLog {
private static final String EXCHANGE_NAME = "logs";
public static void main(String[] args) throws Exception {
//获取连接
Connection cOnnection= ConnectionUtil.getConnection("localhost", 5672, "/", "guest", "guest");
Channel channel = connection.createChannel();
//声明交换机，发布订阅模式
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
//发送消息
for (int i = 0; i <10; i++) {
String message = " message" + i;
System.out.println("[send]：" + message);
//发送消息
channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("utf-8"));
}
channel.close();
connection.close();
}
}

    public class PSReceiveLogs2 {
private static final String Exchange_name = "logs";
public static void main(String[] argv) throws Exception {
//获取连接
Connection cOnnection= ConnectionUtil.getConnection("localhost", 5672, "/", "guest", "guest");
Channel channel = connection.createChannel();
channel.exchangeDeclare(Exchange_name, "fanout");
//随机定义一个队列名称，也可以自己定义一个队列名称
String queueName = channel.queueDeclare().getQueue();
//绑定队列
channel.queueBind(queueName, Exchange_name, "");
DeliverCallback deliverCallback = ((consumerTag, delivery) -> {
String message = new String(delivery.getBody(), "UTF-8");
System.out.println(" [x] Received &＃39;" + message + "&＃39;");
});
channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {
});
}
}

    运行的时候需要首先运行消费者代码，不然没有队列，交换机不知道把消息投递到那些队列中。

  
  

  ? 也可以声明一个非临时队列，只需在绑定队列前面加上如下代码：

  //随机定义一个队列名称,也可以自己定义一个队列名称
// String queueName = channel.queueDeclare().getQueue();
String queueName="A";
//声明队列
channel.queueDeclare(queueName,false,false,false,null);

  其实发布订阅模式就是将消息发送到不同的队列中，由消费者选择不同的队列进行消费即可。




• Springboot实现发布订阅

  @SpringBootApplication
@EnableScheduling
public class RabbitAmqpTutorialsApplication {
public static void main(String[] args) throws Exception {
SpringApplication.run(RabbitAmqpTutorialsApplication.class, args);
}
}

  @Configuration
public class Tut3Config {
/**
* 交换器
* @return
*/
@Bean
public FanoutExchange fanout() {
return new FanoutExchange("tut.fanout");
}
//@Profile("receiver")
private static class ReceiverConfig {
/**
* 队列1
* @return
*/
@Bean
public Queue autoDeleteQueue1() {
return new AnonymousQueue();
}
/**
* 队列2
* @return
*/
@Bean
public Queue autoDeleteQueue2() {
return new AnonymousQueue();
}
/**
* 绑定队列
* @param fanout
* @param autoDeleteQueue1
* @return
*/
@Bean
public Binding binding1(FanoutExchange fanout,
Queue autoDeleteQueue1) {
return BindingBuilder.bind(autoDeleteQueue1).to(fanout);
}
@Bean
public Binding binding2(FanoutExchange fanout,
Queue autoDeleteQueue2) {
return BindingBuilder.bind(autoDeleteQueue2).to(fanout);
}
@Bean
public Tut3Receiver receiver() {
return new Tut3Receiver();
}
}
//@Profile("sender")
@Bean
public Tut3Sender sender() {
return new Tut3Sender();
}
}

  public class Tut3Receiver {
/**
* 监听队列
* @param in
* @throws InterruptedException
*/
@RabbitListener(queues = "#{autoDeleteQueue1.name}")
public void receive1(String in) throws InterruptedException {
receive(in, 1);
}
@RabbitListener(queues = "#{autoDeleteQueue2.name}")
public void receive2(String in) throws InterruptedException {
receive(in, 2);
}
public void receive(String in, int receiver) throws InterruptedException {
StopWatch watch = new StopWatch();
watch.start();
System.out.println("instance " + receiver + " [x] Received &＃39;" + in + "&＃39;");
doWork(in);
watch.stop();
System.out.println("instance " + receiver + " [x] Done in "
+ watch.getTotalTimeSeconds() + "s");
}
private void doWork(String in) throws InterruptedException {
for (char ch : in.toCharArray()) {
if (ch == &＃39;.&＃39;) {
Thread.sleep(1000);
}
}
}
}

  public class Tut3Sender {
@Autowired
private RabbitTemplate template;
@Autowired
private FanoutExchange fanout;
AtomicInteger dots = new AtomicInteger(0);
AtomicInteger count = new AtomicInteger(0);
@Scheduled(fixedDelay = 1000, initialDelay = 500)
public void send() {
StringBuilder builder = new StringBuilder("Hello");
if (dots.getAndIncrement() == 3) {
dots.set(1);
}
for (int i = 0; i builder.append(&＃39;.&＃39;);
}
builder.append(count.incrementAndGet());
String message = builder.toString();
template.convertAndSend(fanout.getName(), "", message);
System.out.println(" [x] Sent &＃39;" + message + "&＃39;");
}
}

  相关代码链接： https://github.com/albert-liu435/springmq