【Rabbitmq】基础工作模式详解

创建rabbitmq连接

public class CreateRabbitmqConnection {
    public static Connection getConnection() throws IOException, TimeoutException {
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("127.0.0.1");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("guest");
        connectionFactory.setPassword("guest");
        connectionFactory.setVirtualHost("/peng");
        return connectionFactory.newConnection();
    }
}

一个生产者对应多个消费者，他们竞争获得消息。

生产者实现：创建一个work_queues的队列，发送10条消息。

        //创建连接
		Connection connection = CreateRabbitmqConnection.getConnection();
        //创建线程
        Channel channel1 = connection.createChannel();
        AMQP.Queue.DeclareOk work_queues = channel1.queueDeclare
                ("work_queues", true, false, false, null);
        for (int i = 0; i < 10; i++) {
         channel1.basicPublish("","work_queues",null,"连接成功");
        }
        channel1.close();
        connection.close();

消费者实现：创建两个线程同时接送work_queues的消息

        Connection connection = CreateRabbitmqConnection.getConnection();
        Channel channel1 = connection.createChannel();
        channel1.queueDeclare("work_queues",true,false,false,null);
        com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel1){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println(consumerName+" : "+new String(body));
            }
        };
        channel1.basicConsume("work_queues",true,consumer);

运行结果

一条消息可以被多个消费者使用。一个消费者监听一个队列，队列中的消息由交换机转发过去。

P：生产者
X：交换机，交换机只负责转发消息，没有存储消息的能力，当没有队列和交换机绑定时，消息会丢失，交换机有三种常见的模型：
·Fanout：广播，将消息转发给所有绑定交换机的队列。
·Direct：定向，将消息转发给符合指定routing key的队列
·Topic：通配符，将消息转发给符合指定routing pattern的队列
C：消费者

生产者代码：创建一个交换机exchange_fanout，为广播类型，绑定队列queue_fanout_1和queue_fanout_2。

 Connection connection = CreateRabbitmqConnection.getConnection();
        Channel channel = connection.createChannel();
        
        String exchangeName = "exchange_fanout";
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT,true,false,false,null);

        //创建队列
        String queueName1 = "queue_fanout_1";
        String queueName2 = "queue_fanout_2";
        channel.queueDeclare(queueName1,true,false,false,null);
        channel.queueDeclare(queueName2,true,false,false,null);

        
        channel.queueBind(queueName1,exchangeName,"");
        channel.queueBind(queueName2,exchangeName,"");

        channel.basicPublish(exchangeName,"",null,"**fanout交换机发送一条消息".getBytes());

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

消费者代码，创建两个消费者，分别绑定queue_fanout_1和queue_fanout_2。

    public static void main(String[] args) throws IOException, TimeoutException {
        Consumer c1 = new Consumer("消费者1号","queue_fanout_1");
        Consumer c2 = new Consumer("消费者2号","queue_fanout_2");
        new Thread(c1).start();
        new Thread(c2).start();

    }

结果

如果此时将queue_fanout_1取消和exchange_fanout的绑定，结果会怎样，在rabbitmq管理网页中操作
此时就只有消费者2号接收到了消息。

生产者代码：创建一个交换机exchange_direct，创建两个队列queue_error和queue_normal，queue_error绑定routingkey（routing_error、routing_data）；queue_normal绑定routingkey（routing_insert、routing_select、routing_data）。

        Connection c = CreateRabbitmqConnection.getConnection();
        Channel channel = c.createChannel();
        //声明交换机
        String exchangeName = "exchange_direct";
        channel.exchangeDeclare(exchangeName,BuiltinExchangeType.DIRECT,true,false,false,null);
        //声明队列,两个队列，一个处理错误消息，一个处理正常消息（正常消息包括，插入、查找），两个队列的消息都要保存到数据库
        String queueName1 = "queue_error";
        String queueName2 = "queue_normal";

        channel.queueDeclare(queueName1,true,false,false,null);
        channel.queueDeclare(queueName2,true,false,false,null);
        //RoutingKey
        String routing_error = "error";
        String routing_insert = "insert";
        String routing_select = "select";
        String routing_data = "data";

        //绑定
        channel.queueBind(queueName1,exchangeName,routing_error);
        channel.queueBind(queueName1,exchangeName,routing_data);
        channel.queueBind(queueName2,exchangeName,routing_insert);
        channel.queueBind(queueName2,exchangeName,routing_select);
        channel.queueBind(queueName2,exchangeName,routing_data);

        
        channel.basicPublish(exchangeName,routing_data,null,
                (routing_data+"操作："+"info").getBytes());
        channel.basicPublish(exchangeName,routing_error,null,
                (routing_error+"操作："+"info").getBytes());
        channel.basicPublish(exchangeName,routing_insert,null,
                (routing_insert+"操作："+"info").getBytes());

        channel.close();
        c.close();

topics通配符模式和routing模式十分相似，只不过前者用的通配符匹配符合要求的routingkey，后者则是将routingkey固定了。