rabbitMQ(二) java简单使用rabbitMQ

亦凉 2022-06-02 05:08 339阅读 0赞

前言：在这里我将用java来简单的实现rabbitMQ。下面我们带着下面问题来一步步的了解和学习rabbitMQ。

1：如果消费者连接中断，这期间我们应该怎么办

2：如何做到负载均衡

3：如何有效的将数据发送到相关的接收者？就是怎么样过滤

4：如何保证消费者收到完整正确的数据

5：如何让优先级高的接收者先收到数据

1)”Hello RabbitMQ”的实现  
用Java编写两个程序; 发送单个消息的生产者，以及接收消息并将其打印出来的消费者。我们将详细介绍Java API中的一些细节，仅仅着眼于这个非常简单的事情，只是为了开始。这是一个消息传递的“Hello World”。  
![这里写图片描述][SouthEast]

send.java

import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    public class Send { 
        private final static String QUEUE_NAME = "hello";
    
          public static void main(String[] argv)
              throws Exception {
              //创建连接工厂
              ConnectionFactory factory = new ConnectionFactory();
              //把工厂和本地机器上的rabbitmq进行绑定
              factory.setHost("localhost");
              //在工厂上拿到连接
              Connection connection = factory.newConnection();
              //获取通道
              Channel channel = connection.createChannel();
              //通道上定义队列
              channel.queueDeclare(QUEUE_NAME, false, false, false, null);
              String message = "hello world";
              //发布消息
              channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
              System.out.println(" [x] Sent '" + message + "'");
              //关闭连接和通道
              channel.close();
              connection.close();
          }
    }

Recv.java:

import java.io.IOException;
    
    import com.rabbitmq.client.AMQP.BasicProperties;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.Consumer;
    import com.rabbitmq.client.DefaultConsumer;
    import com.rabbitmq.client.Envelope;
    
    public class Recv { 
        private final static String QUEUE_NAME = "hello";
    
          public static void main(String[] argv)
              throws Exception {
    
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("localhost");
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
    
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
            //处理消息的对象
            Consumer consumer = new DefaultConsumer(channel) {
    
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope,
                            BasicProperties properties, byte[] body) throws IOException {
                        // TODO Auto-generated method stub
                        String message = new String(body, "UTF-8");
                        System.out.println(" [x] Received '" + message + "'");
                    }
                };
                //把处理消息对象绑定到通道上
                channel.basicConsume(QUEUE_NAME, true, consumer);
            }
    }

运行结果：![这里写图片描述][SouthEast 1]

2）工作队列  
工作队列（又名：任务队列）背后的主要思想是避免立即执行资源密集型任务，必须等待完成。相反，我们安排稍后完成任务。我们把一个任务封装 成一个消息并发送给一个队列。在后台运行的工作进程将弹出任务并最终执行作业。当你运行许多工人时，任务将在他们之间共享。  
![这里写图片描述][SouthEast 2]  
NewTask.java:

import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.MessageProperties;
    
    
    public class NewTask { 
        private static final String  TASK_QUEUE_NAME="task_queue";
    
        public static void main(String[] args) throws Exception {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("localhost");
            Connection connection = connectionFactory.newConnection();//获取connection
            Channel channel = connection.createChannel();//获取管道流
            channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
            /** * queueDeclare(String queue,boolean durable,boolean exclusive,boolean autoDelete,Map<String,Object> arguments)throws IOException queue - the name of the queue durable - true if we are declaring a durable queue (the queue will survive a server restart) exclusive - true if we are declaring an exclusive queue (restricted to this connection) autoDelete - true if we are declaring an autodelete queue (server will delete it when no longer in use) arguments - other properties (construction arguments) for the queue */
            for (int i = 0; i <10; i++) {
                String sendMessage = "send message " +i;
                channel.basicPublish("", TASK_QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, sendMessage.getBytes());
                System.out.println("[S] 发送的信息："+sendMessage);
            }
            channel.close();
            connection.close();
        }
    }

这里需要创建两个work类，代码都是一样的。

import java.io.IOException;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.Consumer;
    import com.rabbitmq.client.DefaultConsumer;
    import com.rabbitmq.client.Envelope;
    import com.rabbitmq.client.AMQP.BasicProperties;
    
    public class Work1 { 
        private static final String  TASK_QUEUE_NAME="task_queue";
    
        public static void main(String[] args) throws  Exception {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("localhost");
            Connection connection = connectionFactory.newConnection();
            final Channel channel = connection.createChannel();
            channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
            channel.basicQos(1);//每次只拿取一次
             final Consumer consumer = new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope,
                        BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body,"utf-8");
                System.out.println("Worker1 Received '" + message + "'");
                try {
                        throw new Exception();
                    } catch (Exception e) {
                    }finally{
                        System.out.println("DO work");
                    }
                }
    
            };
            boolean ack = true;
            channel.basicConsume(TASK_QUEUE_NAME, ack, consumer);
        }
    }

运行结果：  
![这里写图片描述][SouthEast 3]

3）发布和订阅  
工作队列背后的假设是每个任务只被传递给一个工作人员。在这一部分，我们将做一些完全不同的事情 - 我们会向多个消费者传递信息。这种模式被称为“发布/订阅”。  
RabbitMQ中消息传递模型的核心思想是生产者永远不会将任何消息直接发送到队列中。实际上，生产者通常甚至不知道一个消息是否会被传送到任何队列中。

相反，制作人只能发送消息给交易所。交换是一件非常简单的事情。一方面它接收来自生产者的消息，另一方则推动他们排队。交易所必须知道如何处理收到的消息。是否应该附加到特定的队列？是否应该附加到许多队列？还是应该丢弃。这些规则是由交换类型定义的 。  
![这里写图片描述][SouthEast 4]

交换的类型有四种：direct, topic, headers and fanout。这里我们使用fanout.  
临时队列：  
首先，每当我们连接到rabbitMQ，我们需要一个新的，空的队列。要做到这一点，我们可以创建一个随机名称的队列，或者，甚至更好 - 让服务器为我们选择一个随机队列名称。

其次，一旦我们断开消费者，队列应该被自动删除。

在Java客户端中，当我们不给queueDeclare（）提供参数时，我们 用一个生成的名称创建一个非持久的，独占的自动删除队列：

String queue = channel.queueDeclare().getQueue();

绑定：  
我们已经创建了一个扇出交换和一个队列。现在我们需要告诉交换机将消息发送到我们的队列。交换和队列之间的关系被称为绑定。

EmitLog.java:

import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    /** * 发布和订阅，生产者 * @author Administrator * */
    public class EmitLog { 
        private static final String EXCHANGE_NAME ="logs";
    
        public static void main(String[] args) throws Exception {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("localhost");
            Connection connection = connectionFactory.newConnection();
            Channel channel = connection.createChannel();
            //声明通道的交换类型
            channel.exchangeDeclare(EXCHANGE_NAME,BuiltinExchangeType.FANOUT);
            for (int i = 0; i <10; i++) {
                String messageString = "rabbitmq of new "+i;
                //发布
                channel.basicPublish(EXCHANGE_NAME, "", null, messageString.getBytes());
                System.out.println("[S] 发送的东西："+messageString);
            }
            //关闭流
            channel.close();
            connection.close();
        }
    }

处理类有两个，代码都是一样的。  
ReceiveLogs.java

import java.io.IOException;
    
    import com.rabbitmq.client.AMQP.BasicProperties;
    import com.rabbitmq.client.BuiltinExchangeType;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.Consumer;
    import com.rabbitmq.client.DefaultConsumer;
    import com.rabbitmq.client.Envelope;
    
    public class ReceiveLogs {
        private static final String EXCHANGE_NAME ="logs";
    
        public static void main(String[] args) throws Exception {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            connectionFactory.setHost("localhost");
            Connection connection = connectionFactory.newConnection();
            Channel channel = connection.createChannel();
            //声明通道的交换类型
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
            final String queue = channel.queueDeclare().getQueue();
            channel.queueBind(queue, EXCHANGE_NAME, "");//绑定
            Consumer consumer = new DefaultConsumer(channel){
    
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope,
                        BasicProperties properties, byte[] body) throws IOException {
                    System.out.println("[R] 接受的内容："+new String(body, "utf-8")+",queue名"+queue);
                }
    
            };
            channel.basicConsume(queue, true, consumer);
    
        }
    }

结果：  
![这里写图片描述][SouthEast 5]

[SouthEast]: /images/20220602/65e382dbeff64239b1f910070a8cf39d.png
[SouthEast 1]: /images/20220602/68044ad471e14e1fa66048ce430beb9a.png
[SouthEast 2]: /images/20220602/f6486e570ff14d4f915208435d75f1a8.png
[SouthEast 3]: /images/20220602/00888e57602647e4a87b1a06b599075b.png
[SouthEast 4]: /images/20220602/c2f02cd622144bb1a0068145988235fd.png
[SouthEast 5]: /images/20220602/bc187ccb6e21489e9dcd92f85a284fd8.png