RabbitMQ入门案例

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

public class Producer {
    public static void main(String[] args) {
        //1. 创建连接工程
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //这里要使用自己的IP地址
        connectionFactory.setHost("192.168.57.129");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        connectionFactory.setVirtualHost("/");
        Connection connection = null;
        Channel channel = null;
        try {
            //2. 创建连接 connection
            connection = connectionFactory.newConnection("生产者");
            //3. 通过连接获取通道 Channel
            channel = connection.createChannel();
            //4. 通过通道创建交换机，声明队列，绑定关系，路由key，发送消息
            String quequeName = "queuel";
            /**
             * @params1 队列的名称
             * @params2 是否要持久化 durable-false
             * @params3 排他性，是否是独占独立
             * @params4 是否自动删除，随着最后一个消费者消息完毕以后是否把队列自动删除
             * @params5 携带的附属参数
             */
            channel.queueDeclare(quequeName,false,false,false,null);
            //5. 准备消息内容
            String message = "Hello，Consumer";
            //6. 发送消息给队列 queue
            channel.basicPublish("",quequeName,null,message.getBytes());
            System.out.println("消息发送成功");
        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            //7. 关闭连接
            if (channel != null && channel.isOpen()){
                try {
                    channel.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
            //8. 关闭通道
            if (connection != null && connection.isOpen()){
                try {
                    connection.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

public class Consumer {

    public static void main(String[] args) {
        //1. 创建连接工程
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //这里要使用自己的IP地址
        connectionFactory.setHost("192.168.57.129");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
        connectionFactory.setVirtualHost("/");

        Connection connection = null;
        Channel channel = null;
        try {
            //2. 创建连接 connection
            connection = connectionFactory.newConnection("消费者");
            //3. 通过连接获取通道 Channel
            channel = connection.createChannel();
            //4. 通过通道创建交换机，声明队列，绑定关系，路由key，发送消息，和接收消息
            String quequeName = "queue1";
            channel.queueDeclare(quequeName,false,false,false,null);
            //5.监听消息
            DefaultConsumer consumer = new DefaultConsumer(channel){
                /*
                consumerTag：消息者标签，channel.basicConsume可以指定
                envelope：消息包内容，可从中获取消息id，消息routing key，交换机，消息和重装标记（收到消息失败后是否需要重新发送）
                properties：消息属性
                body；消息
                */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    //路由key
                    System.out.println("路由key为："+ envelope.getRoutingKey());
                    //交换机
                    System.out.println("交换机为："+ envelope.getExchange());
                    //消息id
                    System.out.println("消息id为："+ envelope.getDeliveryTag());
                    //收到的消息
                    System.out.println("接收到的消息："+ new String(body,"UTF-8"));
                    System.out.println("");
                    System.out.println("======================================================");
                    System.out.println("");
                }
            };
            channel.basicConsume("queue1", true, consumer);
        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            //6. 不关闭资源，一直监听
        }
    }
}