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

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);
    }
}

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

.

exchang：