RabbitMQ订阅发布模式

• 首先我们需要一两个消费者才能展示我们的订阅发布模式
• • 首先先展示我的工具类——也就是获取信道的工具类。
  • 消费者类
  • 写完我们的消费者，就是我们的生产着了
  • 最后我们就可以展示我们的订阅发布模式了。

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;


public class RabbitMQUtils {

    public static Channel getChannel()throws Exception{
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("*.*.*.*");//你自己的服务器IP地址
        factory.setUsername("guest");//用户名
        factory.setPassword("guest");//密码
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        return channel;
    }

}

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.yyp.rabbitmq.utils.RabbitMQUtils;


public class ReceiveLogs1 {

    //交换机名称
    public static final String EXCHANGE_NAME = "logs";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtils.getChannel();
        //声明交换机
        channel.exchangeDeclare(EXCHANGE_NAME,"fanout");
        
        String queuename = channel.queueDeclare().getQueue();
        
        channel.queueBind(queuename,EXCHANGE_NAME,"");
        System.out.println("1等待就收消息，把接收到消息打印在控制台上。。。。。。");
        DeliverCallback deliverCallback = (consumerTag,message) ->{
            System.out.println(new String(message.getBody()));
        };
        channel.basicConsume(queuename,true,deliverCallback,(c,m) ->{});
    }

}

我们需要两个消费者

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.yyp.rabbitmq.utils.RabbitMQUtils;


public class ReceiveLogs2 {

    //交换机名称
    public static final String EXCHANGE_NAME = "logs";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtils.getChannel();
        //声明交换机
        channel.exchangeDeclare(EXCHANGE_NAME,"fanout");
        
        String queuename = channel.queueDeclare().getQueue();
        
        channel.queueBind(queuename,EXCHANGE_NAME,"");
        System.out.println("2等待就收消息，把接收到消息打印在控制台上。。。。。。");
        DeliverCallback deliverCallback = (consumerTag,message) ->{
            System.out.println(new String(message.getBody()));
        };
        channel.basicConsume(queuename,true,deliverCallback,(c,m) ->{});
    }

}

import com.rabbitmq.client.Channel;
import com.yyp.rabbitmq.utils.RabbitMQUtils;

import java.util.Scanner;


public class EmitLog {

    public static final String EXCHANGE_NAME = "logs";

    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtils.getChannel();
        channel.exchangeDeclare(EXCHANGE_NAME,"fanout");
        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNext()){
            String message = scanner.next();
            channel.basicPublish(EXCHANGE_NAME,"",null,message.getBytes());
            System.out.println("已发送消息");
        }
    }

}

我们消息发送了，两个都能接收，这就是订阅发布模式