RabbitMQ的Java调用实战

亦凉 2023-10-01 11:27 3阅读 0赞

本文将介绍RabbitMQ的Java调用代码。

# 1、Jar包依赖 #

新建一个maven工程，添加amqp-client依赖

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

# 2、连接工具类 #

关键点如下：

创建ConnectionFactory，包括设置rabbitmq服务的地址与端口号（amqp默认端口是5672）、用户名、密码、vhost(默认是/) ;

代码实例如下：

public class ConnectionUtil {
        /**
         * 建立与RabbitMQ的连接
         * @return
         * @throws Exception
         */
        public static Connection getConnection() throws Exception {
            //定义连接工厂
            ConnectionFactory factory = new ConnectionFactory();
            //设置服务地址
            factory.setHost("192.168.254.191");
            //端口
            factory.setPort(5672);
            //设置账号信息，用户名、密码、vhost
            factory.setVirtualHost("/vhost");//设置虚拟机，一个mq服务可以设置多个虚拟机，每个虚拟机就相当于一个独立的mq
            factory.setUsername("summer");
            factory.setPassword("123456");
            // 通过工厂获取连接
            Connection connection = factory.newConnection();
            return connection;
        }
    }

# 3、消息生产者 #

关键点如下：

1、获取Connection ;

2、获取Channel ，并在其上声明queue、exchanger，并通过routingkey把queue在exchanger进行绑定 ;

代码实例如下：

public class Producer {
     
        private final static String QUEUE_NAME = "simple_queue";
     
        public static void main(String[] argv) throws Exception {
            // 1、获取到连接
            Connection connection = ConnectionUtil.getConnection();
            // 2、从连接中创建通道，使用通道才能完成消息相关的操作
            Channel channel = connection.createChannel();
            // 3、声明（创建）队列
            //参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
            /**
             * 参数明细
             * 1、queue 队列名称
             * 2、durable 是否持久化，如果持久化，mq重启后队列还在
             * 3、exclusive 是否独占连接，队列只允许在该连接中访问，如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建
             * 4、autoDelete 自动删除，队列不再使用时是否自动删除此队列，如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）
             * 5、arguments 参数，可以设置一个队列的扩展参数，比如：可设置存活时间
             */
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            // 4、消息内容
            String message = "Hello World!";
            // 向指定的队列中发送消息
            //参数：String exchange, String routingKey, BasicProperties props, byte[] body
            /**
             * 参数明细：
             * 1、exchange，交换机，如果不指定将使用mq的默认交换机（设置为""）
             * 2、routingKey，路由key，交换机根据路由key来将消息转发到指定的队列，如果使用默认交换机，routingKey设置为队列的名称
             * 3、props，消息的属性
             * 4、body，消息内容
             */
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println(" [x] Sent '" + message + "'");
            
            //关闭通道和连接(资源关闭最好用try-catch-finally语句处理)
            channel.close();
            connection.close();
        }
    }

# 4、消息消费者 #

关键点如下：

1、获取Connection ;

2、获取Channel ,并在其上声明queue ;

3、定义消费消息Consumer；

4、消费消息channel.basicConsume(queueName,false,"queueOne",consumer);

public class Consumer{
        private final static String QUEUE_NAME = "simple_queue";
     
        public static void main(String[] argv) throws Exception {
            // 获取到连接
            Connection connection = ConnectionUtil.getConnection();
            //创建会话通道,生产者和mq服务所有通信都在channel通道中完成
            Channel channel = connection.createChannel();
            // 声明队列
            //参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
            /**
             * 参数明细
             * 1、queue 队列名称
             * 2、durable 是否持久化，如果持久化，mq重启后队列还在
             * 3、exclusive 是否独占连接，队列只允许在该连接中访问，如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建
             * 4、autoDelete 自动删除，队列不再使用时是否自动删除此队列，如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）
             * 5、arguments 参数，可以设置一个队列的扩展参数，比如：可设置存活时间
             */
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            //实现消费方法
            DefaultConsumer consumer = new DefaultConsumer(channel){
                // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
                /**
                 * 当接收到消息后此方法将被调用
                 * @param consumerTag  消费者标签，用来标识消费者的，在监听队列时设置channel.basicConsume
                 * @param envelope 信封，通过envelope
                 * @param properties 消息属性
                 * @param body 消息内容
                 * @throws IOException
                 */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    //交换机
                    String exchange = envelope.getExchange();
                    //消息id，mq在channel中用来标识消息的id，可用于确认消息已接收
                    long deliveryTag = envelope.getDeliveryTag();
                    // body 即消息体
                    String msg = new String(body,"utf-8");
                    System.out.println(" [x] received : " + msg + "!");
                }
            };
            
            // 监听队列，第二个参数：是否自动进行消息确认。
            //参数：String queue, boolean autoAck, Consumer callback
            /**
             * 参数明细：
             * 1、queue 队列名称
             * 2、autoAck 自动回复，当消费者接收到消息后要告诉mq消息已接收，如果将此参数设置为tru表示会自动回复mq，如果设置为false要通过编程实现回复
             * 3、callback，消费方法，当消费者接收到消息要执行的方法
             */
            channel.basicConsume(QUEUE_NAME, true, consumer);
        }
    }