一、概述

前面学过ActiveMQ。ActiveMQ主要是实现的JMS规范，而RabbitMQ就是AMQP的一个具体实现。

RabbitMQ里面有几个概念：生产者、消费者、消息、交换器、路由键、队列、绑定、虚拟主机

1.生产者角度

　　生产者产生数据，然后根据指定交换器和路由键将数据发送到消息队列RabbitMQ。为了保证交换器的存在，我们每次在初始化生产者的时候都要尝试去创建一个交换器。

　　交换器总共有4种类型：

direct 路由键完全匹配
fanout 消息广播，将忽略路由键
topic 通过“*”和“#”的通配符进行绑定。注意：”.”将路由键分为了几个标识符，“*”匹配1个，“#”匹配一个或多个
headers 和direct类似，很少使用

2.消费者角度

　　消费者主要就是获取并消费数据，因此需要创建一个队列，同时需要创建一个交换器（交换器在消费者和生产者都可以创建），然后将队列和交换器通过路由键进行绑定。最后就可以根据队列进行数据的消费了。

二、Java代码

1.pom.xml

<dependency>

    <groupId>com.rabbitmq</groupId>

    <artifactId>amqp-client</artifactId>

    <version>4.0.2</version>

    <exclusions>

        <exclusion>

            <groupId>org.slf4j</groupId>

            <artifactId>slf4j-api</artifactId>

        </exclusion>

    </exclusions>

</dependency>

2.生产者代码

package cn.duanjt;

import java.io.IOException;

import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.BuiltinExchangeType;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.ConnectionFactory;

/**

 * 生产者

 * @author 段江涛

 * @date 2018-11-30

 */

public class Productor {

    public static void main(String[] args) throws IOException, TimeoutException {

        String ROUTE_KEY = "rabbitmq-duanjt";// 路由键名称

        String EXCHANGE_NAME = "exchange-duanjt";// 交换器名称

        ConnectionFactory factory = new ConnectionFactory();

        factory.setHost("192.168.23.24");

        factory.setPort(5672);

        factory.setUsername("admin");

        factory.setPassword("admin");

        factory.setVirtualHost("/");//虚拟主机，可通过控制台查看

        //创建连接和信道

        Connection conn = factory.newConnection();

        Channel channel = conn.createChannel();

        // 创建一个交换器，参数为：交互器名称和交换器类型

        // 注意：其实这个交换器只需要声明一次就可以，但是由于无法保证交换器已经存在了，所以我们每次都要声明

        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);

        for (int i = 0; i < 5; i++) {

            String msg = "Hello world.I love you forever ===>" + i;

            // 发布消息，需要参数：交换器，路由键。最后一个参数为消息内容

            // 注意：RabbitMQ的消息类型只有一种，那就是byte[]

            channel.basicPublish(EXCHANGE_NAME, ROUTE_KEY, null, msg.getBytes("utf-8"));

            System.out.println("send:" + msg);

        }

        //关闭信道和连接

        channel.close();

        conn.close();

    }

}

3.消费者代码

package cn.duanjt;

import java.io.IOException;

import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.Channel;

import com.rabbitmq.client.Connection;

import com.rabbitmq.client.ConnectionFactory;

import com.rabbitmq.client.DefaultConsumer;

import com.rabbitmq.client.Envelope;

import com.rabbitmq.client.AMQP.BasicProperties;

import com.rabbitmq.client.BuiltinExchangeType;

public class Consumer {

    public static void main(String[] args) throws IOException, TimeoutException {

        String QUEUE_NAME = "queue-duanjt";// 队列名称

        String ROUTE_KEY = "rabbitmq-duanjt";// 路由键名称

        String EXCHANGE_NAME = "exchange-duanjt";// 交换器名称

        ConnectionFactory factory = new ConnectionFactory();

        factory.setHost("192.168.23.24");

        factory.setPort(5672);

        factory.setUsername("admin");

        factory.setPassword("admin");

        factory.setVirtualHost("/");

        Connection conn = factory.newConnection();

        Channel channel = conn.createChannel();

        // 创建一个队列

        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 创建交换器

        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);

        // 将队列和交换器通过路由键进行绑定

        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTE_KEY);

        //开始消费，第二个参数表示自动确认

        channel.basicConsume(QUEUE_NAME, true, new DefaultConsumer(channel) {

            // 当消息到达时执行回调方法

            @Override

            public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body) throws IOException {

                String message = new String(body, "utf-8");

                System.out.println("[Receive]：" + message);

            }

        });

    }

}

注意：

1.为了保证交换器的存在，所以消费者和生产者都要创建，因为不知道是消费者先启动还是生产者先启动

2.可以通过http://ip:15672 查看交换器、路由键和队列之间的关系

3.一个连接(Connection)可以创建多个信道(Channel)。每个信道也可以在独立的一个线程里面

4.一个队列可以有多个消费者，这种情况下，消息将在消费者之间进行轮询

巴特西

Java 访问RabbitMQ

一、概述

二、Java代码

最新文章

热门文章