三RabbitMQ中模式—简单模式Hello World

Posted 2021-10-27 上善若水

一、Hello World程序

我们将用Java编写两个程序，发送单个消息的生产者和接收消息并打印出来的消费者。将介绍Java API中的一些细节。
在下图中，"P"是我们的生产者，"C"是我们的消费者。中间的框是一个队列——RabbitMQ代表使用者保留的消息缓冲区。

1.1、加入依赖

    <dependencies>
        <!-- rabbitmq依赖客户端 -->
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>5.8.0</version>
        </dependency>
        <!-- 操作文件流的一个依赖 -->
        <dependency>
            <groupId>commons-io</groupId>
            <artifactId>commons-io</artifactId>
            <version>2.6</version>
        </dependency>
    </dependencies>

    <!-- 指定jdk编译版本-->
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>8</source>
                    <target>8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>

1.2、消息生产者

package com.xbmu.rabbitmq.one;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

/**
 * 消息生产者
 */
public class Producer {
    private final static String QUEUE_NAME = "hello";

    public static void main(String[] args) {
        // 创建一个连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        // 设置工厂IP,连接rabbitmq的队列
        factory.setHost("192.168.184.110");
        // 用户名
        factory.setUsername("admin");
        // 密码
        factory.setPassword("123");

        try {
            // 创建连接
            Connection connection = factory.newConnection();
            // 从连接中获取一个通道
            // channel 实现了自动close接口 自动关闭 不需要显示关闭
            Channel channel = connection.createChannel();
            /**
             * Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,
             *                                  Map<String, Object> arguments) throws IOException;
             * 生成一个队列
             * 1.队列名称
             * 2.队列里面的消息是否持久化 默认消息存储在内存中
             * 3.该队列是否只供一个消费者进行消费 是否进行共享 true可以多个消费者消费
             * 4.是否自动删除 最后一个消费者断开连接以后 该队列是否自动删除  true自动删除
             * 5.其他参数
             *
             */
            channel.queueDeclare(QUEUE_NAME,false,false,false,null);
            // 发消息
            String message = "hello world";
            /**
             * void basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body) throws IOException;
             * 发送一个消息
             * 1.发送到哪个交换机
             * 2.路由的key是哪个,本次是队列的名称
             * 3.其他的参数消息
             * 4.发送消息的消息体
             */
            channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
            System.out.printf("消息发送完毕");
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}

1.3、消息消费者

package com.xbmu.rabbitmq.one;

import com.rabbitmq.client.*;

public class Consumer {
    private final static String QUEUE_NAME = "hello";

    public static void main(String[] args) {
        // 创建一个连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        // 设置工厂IP,连接rabbitmq的队列
        factory.setHost("192.168.184.110");
        // 用户名
        factory.setUsername("admin");
        // 密码
        factory.setPassword("123");
        try {
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();
            System.out.println("等待接收消息......");
            // 推送的消息如何进行消费的接口回调
            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                String message = new String(delivery.getBody());
                System.out.println(message);
            };
            // 取消消费的一个回调接口 如在消费的时候队列被删除掉了
            CancelCallback cancelCallback = (consumerTag) -> {
                System.out.println("消息消费被中断");
            };
            /**
             * 消费者消费消息
             * 1. 消费哪个队列
             * 2. 消费成功之后是否要自动应答 true 代表自动应答 false 手动应答
             * 3. 消费者成功消费的回调
             * 4. 消费者取消消费的回调
             */
            channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}

1.4、测试

首先确保，rabbitmq所在的Linux服务器上，确保：
5672：client端通信端口、15672：管理界面ui端口，开放。

执行生产者类

此时，MQ中产生了一个消息队列，消息是存放在队列中的，队列名为"hello"。


执行消费者类，去消费生产者产生的消息。输出了消息的信息"hello world"。