RabbitMQ之简单队列

刺骨的言语ヽ痛彻心扉 2022-03-01 07:28 261阅读 0赞

## 1. 简单队列图示 ##

![在这里插入图片描述][20190321230914214.png]  
P：消息的生产者  
C : 消息的消费者  
红色 : 队列  
生产者将消息"hello"发送至队列，消费者从队列中获取消息

## 2.导入RabbitMQ依赖包 ##

<dependency>
    	<groupId>com.rabbitmq</groupId>
    	<artifactId>amqp-client</artifactId>
    	<version>3.4.1</version>
    </dependency>

## 3. MQ连接工具类 ##

public class ConnectionUtil {
    
        public static Connection getConnection() throws Exception {
            //定义连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            //设置服务地址
            factory.setHost("localhost");
            //端口
            factory.setPort(5672);
            //设置账号信息，用户名、密码、vhost
            factory.setVirtualHost("/mlz");
            factory.setUsername("admin");
            factory.setPassword("admin");
            // 通过工程获取连接
            Connection connection = factory.newConnection();
            return connection;
        }
    
    }

## 4.生产者生产消息 ##

public class Send {
    
        private final static String QUEUE_NAME = "test_queue";
    
        public static void main(String[] argv) throws Exception {
            try {
                // 获取到连接以及mq通道
                Connection connection = ConnectionUtil.getConnection();
                // 从连接中创建通道
                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();
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
        }
    }

## 5.消费者消费消息 ##

public class Recv {
    
        private final static String QUEUE_NAME = "test_queue";
    
        public static void main(String[] argv) throws Exception {
    
            try {
                //获取到连接以及mq通道
                Connection connection = ConnectionUtil.getConnection();
    
                //创建频道
                Channel channel = connection.createChannel();
    
                //定义队列的消费者
                Consumer consumer = new DefaultConsumer(channel) {
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
                            throws IOException {
                        String message = new String(body, "UTF-8");
                        System.out.println("接收到一个消息： " + message);
                    }
                };
                // 监听队列
                /*
                 true:表示自动确认，只要消息从队列中获取，无论消费者获取到消息后是否成功消费，都会认为消息已经成功消费
                 false:表示手动确认，消费者获取消息后，服务器会将该消息标记为不可用状态，等待消费者的反馈，
                        如果消费者一直没有反馈，那么该消息将一直处于不可用状态，并且服务器会认为该消费者已经挂掉，不会再给其
                        发送消息，直到该消费者反馈。
                */
                channel.basicConsume(QUEUE_NAME, true, consumer);
    
            } catch (IOException e) {
                e.printStackTrace();
            } catch (TimeoutException e) {
                e.printStackTrace();
            }
        }
    }

## 6.简单队列的不足 ##

1.  耦合性高，生产者一 一对应消费者(如果我想有多个消费者消费队列中的消息，就不支持)
2.  队列名变更，就得同时变更

[20190321230914214.png]: /images/20220301/595be4c3cfe94111ab245c534e280c6c.png