RabbitMQ 发布消息，消费消息代码实现

Bertha 。 2022-04-12 05:43 196阅读 0赞

### RabbitMQ 发布消息，消费消息 ###

*   *  pom依赖
     *  生产者
     *  消费者
     *  运行生产者通过管理界面看见效果

RabbitMQ 消息中间件原理 ，介绍请看上一篇博文，这里只写基本实现  
spring有集成rabbitmq，本文未实现spring方式

## pom依赖 ##

导入依赖：

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

## 生产者 ##

public class Producer {
        private final static String QUEUE_NAME = "myQueue";
        private final static String EXCHANGE_NAME = "myExchange";
        private final static String ROUTINGKEY = "myKey";
        /**
         *     TYPE : direct,fanout,topic,headers
         */
        private final static String EXCHANGE_TYPE = "fanout";
    
        public static void main(String[] args) {
    
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("localhost");
            factory.setPort(5672);
            factory.setUsername("guest");
            factory.setPassword("guest");
    
            Connection connection;
            Channel channel;
            try {
                connection = factory.newConnection();
                channel = connection.createChannel();
                channel.exchangeDeclare(EXCHANGE_NAME, EXCHANGE_TYPE);
                channel.queueDeclare(QUEUE_NAME, false, false, false, null);
                channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTINGKEY);
    
                String message = "Say love me";
    
                channel.basicPublish(EXCHANGE_NAME, ROUTINGKEY, null, message.getBytes());
                System.out.println(" [x] Sent '" + message + "'");
                channel.close();
                connection.close();
            }catch (TimeoutException e){
                e.printStackTrace();
            }catch (IOException e){
                e.printStackTrace();
            }
    
        }
    }

## 消费者 ##

public class Customer {
        private final static String QUEUE_NAME = "myQueue";
    
        public static void main(String[] args) throws IOException, TimeoutException {
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("localhost");
            factory.setPort(5672);
            factory.setUsername("guest");
            factory.setPassword("guest");
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            System.out.println("Customer Waiting Received messages");
            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("Customer Received '" + message + "'");
                }
            };
            channel.basicConsume(QUEUE_NAME, true, consumer);
        }
    }

## 运行生产者通过管理界面看见效果 ##

![消息队列详情][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ0OTI2NjU_size_16_color_FFFFFF_t_70].

exchang：  
![exchange详情页面][exchange]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTQ0OTI2NjU_size_16_color_FFFFFF_t_70]: /images/20220412/b9f75e3fef504d18a586f6f1e86d864f.png
[exchange]: /images/20220412/008d5d7cb38a42f0bc18baa8e17c11af.png