天天看点
返回

RabbitMQ介绍及各个工作模式的使用

  1. RabbitMQ使用示例代码
  2. 消息队列(Message Queue)：是一种跨进程的通信机制，主要用于上下文传递消息。
  3. MQ作为消息中间件，最主要的作用是对系统之间传递消息进行"解耦"，MQ是数据可靠性的重要保障。
  4. MQ主要角色是消息代理服务器。
  5. RabbitMQ的优势
    1. RabbitMQ是世界上最火的开源消息代理服务器；
    2. RabbitMQ几乎支持所有的操作系统与编程语言；
    3. RabbitMQ提供了高并发、高可用的成熟方案，支持多种消息协议，易于部署与使用。
  6. RabbitMQ的应用场景
    1. 异构系统的数据传递
    2. 高并发程序的流量控制
    3. 基于P2P的程序
    4. 分布式系统的事务一致性
    5. 高可靠性的交易系统
  7. RabbitMQ常用命令 
    rabbitmq-server  #前台启动服务
rabbitmq-server -detached #后台启动服务
rabbitmqctl stop #停止服务，表示停止进程
rabbitmqctl start_app  #启动应用
rabbitmqctl stop_app   #终止应用，不会关闭进程
rabbitmqctl add_user username password  #创建用户
rabbitmqctl delete_user   #删除用户
rabbitmqctl change_password username newpassword  #修改用户的密码
rabbitmqctl set_user_tags username tag   #授予用户角色(tag)
rabbitmqctl set_permissions -p / username '.*' '.*' '.*'  #设置用户允许访问的vhost(虚拟主机),分别是配置权限、读权限、写权限
  8. 查看网络端口号 
    netstat -tulpn
  9. RabbitMQ的4中tag

    1. 超级管理员administrator：

      可登录管理控制台，可查看所有的信息，并且对用户、策略(policy)进行操作

    2. 监控者monitoring

      登录管理控制台，同时可以查看rabbitmq节点相关的信息(进程数、内存使用情况、磁盘使用情况等)，但无法进行策略的定制

    3. 策略制定者policymaker

      可登录管理控制台，同时可以对policy进行管理，但无法查看节点(实例)相关的信息

    4. 普通管理者management

      仅可登录管理控制台，无法看到节点信息，也无法对策略进行管理

  10. RabbitMQ的guest用户只能在本地登录，如果需要远程登录RabbitMQ控制台，需要新创建一个用户，并将该用户的权限设置为administrator
  11. AMPQ：高级消息队列协议，一个统一提供消息服务的应用层标准高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。基于此协议的客户端与中间件可传递消息，并不受客户端/中间件不同产品、不同的开发语言等条件的限制。Erlang中的实现有RabbitMQ等。

  12. 基本概念

    Producer 生产者，消息的提供者、

    Consumer 消费者，消息的使用者

    Message 消息，程序间的通信数据

    Queue 队列，消息存放的容器，消息先进先出

    Vhost 虚拟主机，相当于MQ的"数据库"，用于存储队列

  13. 消息的状态
    1. Ready:消息已经被传入队列，等待被消费

    2. Unacked：

      消息已经被消费者认领，但还未被确认"已被消费"

      Unacked状态下消费者断开连接，则消息回到Ready

      没有确认，客户端没有断开，则一直处于Unacked

    3. Finished：调用basicAck()方法后，表示消息已经被消费，从队列中移除
  14. 第一次连接RabbitMQ
  15. 导入依赖 
    <dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.6.0</version>
</dependency>
  16. RabbitMQ生产者的代码 
    import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Procuder {

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

        //ConnectionFactory创建MQ的物理连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("127.0.0.1");  //ip地址
        connectionFactory.setPort(5672);         //端口
        connectionFactory.setUsername("guest");  //用户名
        connectionFactory.setPassword("guest");  //密码
        connectionFactory.setVirtualHost("/test"); //虚拟主机

        //TCP物理连接
        Connection connection = connectionFactory.newConnection();

        //创建通信通道，相当于TCP的虚拟连接
        Channel channel = connection.createChannel();

        //创建队列，声明并创建一个队列，如果队列已经存在，则使用这个队列
        //第一个参数，对列名称
        //第二个参数，是否持久话，false表示不持久化数据，MQ停掉后数据就会丢失
        //第三个参数，是否队列私有化，false表示所有的消费者都可以访问，true表示只有第一次拥有它的消费者才可以一直使用，其他消费者不能访问
        //第四个参数，是否自动删除，false连接停掉后不自动删除掉这个队列
        //第五个参数，其他额外的参数
        channel.queueDeclare("helloworld", false, false, false, null);

        //需要发送的消息
        String content = "Hello World bb!";

        //第一个参数，交换机
        //第二个参数，队列名称
        //第三个参数，额外的设置属性
        //第四个参数，需要发送的消息的字节数组
        channel.basicPublish("", "helloworld", null, content.getBytes());

        channel.close();
        connection.close();

        System.out.println("数据发送成功");
    }
}
  17. RabbitMQ消费者的代码 
    import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer {

    public static void main(String[] args) throws IOException, TimeoutException {
        //ConnectionFactory创建MQ的物理连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("127.0.0.1");  //ip地址
        connectionFactory.setPort(5672);         //端口
        connectionFactory.setUsername("guest");  //用户名
        connectionFactory.setPassword("guest");  //密码
        connectionFactory.setVirtualHost("/test"); //虚拟主机

        //TCP物理连接
        Connection connection = connectionFactory.newConnection();

        //创建通道
        Channel channel = connection.createChannel();

        //绑定消息队列
        //第一个参数，对列名称
        //第二个参数，是否持久话，false表示不持久化数据，MQ停掉后数据就会丢失
        //第三个参数，是否队列私有化，false表示所有的消费者都可以访问，true表示只有第一次拥有它的消费者才可以一直使用，其他消费者不能访问
        //第四个参数，是否自动删除，false连接停掉后不自动删除掉这个队列
        //第五个参数，其他额外的参数
        channel.queueDeclare("helloworld", false, false, false, null);

        //创建一个消息消费者
        //第一个参数，队列名称
        //第二个参数，是否自动确认收到消息，false表示手动编写程序来确认消息，这是MQ推荐的做法
        //第三个参数，DefaultConsumer的实现类
        channel.basicConsume("helloworld", false, new Receiver(channel));

        //在消费者中不能关闭channel和connection
    }
}

class Receiver extends DefaultConsumer{

    private Channel channel;

    //重写构造函数，channel通道对象需要从外部传入，在handleDelivery中会用到
    public Receiver(Channel channel) {
        super(channel);
        this.channel = channel;
    }

    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        /*super.handleDelivery(consumerTag, envelope, properties, body);*/

        String messageBody = new String(body);
        System.out.println("消费者接收到： " + messageBody);

        //签收消息，确认消息
        //第一个参数，envelope.getDeliveryTag()获取这个消息的TagId，是一个整数
        //第二个参数，false只确认签收当前的消息，true时，表示签收该消费者所有未签收的消息
        channel.basicAck(envelope.getDeliveryTag(), false);
    }

}
  18. RabbitMQ连接工具类的封装 
    import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class RabbitMQUtil {

    private static ConnectionFactory connectionFactory;

    static {
        //ConnectionFactory创建MQ的物理连接
        connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("127.0.0.1");  //ip地址
        connectionFactory.setPort(5672);         //端口
        connectionFactory.setUsername("guest");  //用户名
        connectionFactory.setPassword("guest");  //密码
        connectionFactory.setVirtualHost("/test"); //虚拟主机
    }

    public static Connection getConnection() {
        Connection connection = null;
        try {
            connection = connectionFactory.newConnection();
        } catch (Exception e) {
            throw new RuntimeException(e);  //不需要显式的声明抛出
        }
        return connection;
    }
}
  19. 优化后的生产者 
    import com.kangswx.rabbitmq.utils.RabbitMQConsts;
import com.kangswx.rabbitmq.utils.RabbitMQUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Procuder {

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

        //TCP物理连接
        Connection connection = RabbitMQUtil.getConnection();

        //创建通信通道，相当于TCP的虚拟连接
        Channel channel = connection.createChannel();

        //创建队列，声明并创建一个队列，如果队列已经存在，则使用这个队列
        //第一个参数，对列名称  helloworld
        //第二个参数，是否持久话，false表示不持久化数据，MQ停掉后数据就会丢失
        //第三个参数，是否队列私有化，false表示所有的消费者都可以访问，true表示只有第一次拥有它的消费者才可以一直使用，其他消费者不能访问
        //第四个参数，是否自动删除，false连接停掉后不自动删除掉这个队列
        //第五个参数，其他额外的参数
        channel.queueDeclare(RabbitMQConsts.QUEUE_HELLO, false, false, false, null);

        //需要发送的消息
        String content = "Hello World bb!";

        //第一个参数，交换机
        //第二个参数，队列名称  helloworld
        //第三个参数，额外的设置属性
        //第四个参数，需要发送的消息的字节数组
        channel.basicPublish("", RabbitMQConsts.QUEUE_HELLO, null, content.getBytes());

        channel.close();
        connection.close();

        System.out.println("数据发送成功");
    }
}
  20. 优化后的消费者 
    import com.kangswx.rabbitmq.utils.RabbitMQConsts;
import com.kangswx.rabbitmq.utils.RabbitMQUtil;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer {

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

        //TCP物理连接
        Connection connection = RabbitMQUtil.getConnection();

        //创建通道
        Channel channel = connection.createChannel();

        //绑定消息队列
        //第一个参数，对列名称  helloworld
        //第二个参数，是否持久话，false表示不持久化数据，MQ停掉后数据就会丢失
        //第三个参数，是否队列私有化，false表示所有的消费者都可以访问，true表示只有第一次拥有它的消费者才可以一直使用，其他消费者不能访问
        //第四个参数，是否自动删除，false连接停掉后不自动删除掉这个队列
        //第五个参数，其他额外的参数
        channel.queueDeclare(RabbitMQConsts.QUEUE_HELLO, false, false, false, null);

        //创建一个消息消费者
        //第一个参数，队列名称  helloworld
        //第二个参数，是否自动确认收到消息，false表示手动编写程序来确认消息，这是MQ推荐的做法
        //第三个参数，DefaultConsumer的实现类
        channel.basicConsume(RabbitMQConsts.QUEUE_HELLO, false, new Receiver(channel));

        //在消费者中不能关闭channel和connection
    }
}

class Receiver extends DefaultConsumer{

    private Channel channel;

    //重写构造函数，channel通道对象需要从外部传入，在handleDelivery中会用到
    public Receiver(Channel channel) {
        super(channel);
        this.channel = channel;
    }

    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        /*super.handleDelivery(consumerTag, envelope, properties, body);*/

        String messageBody = new String(body);
        System.out.println("消费者接收到： " + messageBody);

        //签收消息，确认消息
        //第一个参数，envelope.getDeliveryTag()获取这个消息的TagId，是一个整数
        //第二个参数，false只确认签收当前的消息，true时，表示签收该消费者所有未签收的消息
        channel.basicAck(envelope.getDeliveryTag(), false);
    }

}
  21. RabbitMQ的工作模式
    1. Hello World：最简单的模式，一个生产者对应一个消费者
    2. Work queues：一个生产者生产的消息由多个消费者进行消费，一个消息只能被一个消费者处理
    3. Publish/Subsribe：多了一个交换器，交消息按照一定的规则分发给多个消费者，每个消费者收到的消息是一模一样的，一个消息可能被多个消费者处理
    4. Routing模式：交消息有选择的分发给不同的消费者，不同的消费者收到的消息有可能不一样，缺点是，需要对数据进行精准匹配
    5. Topics：与Routing类似，可以定义一个表达式规则，模糊匹配
    6. RPC：远程调用

  22. Work queues工作队列：

    创建一个工作队列，它会发送一些耗时的任务给多个消费者。

    有多个消息的情况下，Work queue会将消息分发给不同的消费者，每个消费者都会接收到不同的消息，并且可以根据处理消息的速度来接收消息的数量，让消费者程序发挥最大性能。

    特别适合在集群环境中做异步处理，能最大程度的发挥每一台服务器的性能。

  23. 将消费者的处理方式修改为处理完一条，再去队里里面拿下一条数据的方法 
    channel.basicQos(1);  //处理完一条消息再去队列里面取下一条消息
channel.basicAck(envelope.getDeliveryTag(), false);  //调用basicAck的时候，才会去取下一条的数据

  24. Publish/Subsribe模式

    发布/订阅模式中，生产者不再直接与队列进行绑定，而是将数据发送至"交换机Exchange"

    交换机Exchange用于将数据按某种规则送入与之绑定的队列，进而供消费者使用。

    发布/订阅模式中，交换机将无差别的将所有的消息送入与之绑定的队列，所有的消费者拿到的消息完全相同，交换机的类型被称为fanout。

  25. 使用场景：发布订阅模式因为所有消费者获得相同的消息，所以特别适合"数据提供商与应用商"
  26. 创建一个发布订阅模式下的交换机(类型为fanout)
    RabbitMQ介绍及各个工作模式的使用
  27. 生产者代码示例 
    import com.kangswx.rabbitmq.utils.RabbitMQConsts;
import com.kangswx.rabbitmq.utils.RabbitMQUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

import java.io.IOException;
import java.util.Scanner;
import java.util.concurrent.TimeoutException;

/**
 * 气象局(模拟生产者)
 */
public class WeatherBureau {

    public static void main(String[] args) throws IOException, TimeoutException {
        //TCP物理连接
        Connection connection = RabbitMQUtil.getConnection();

        //创建通信通道，相当于TCP的虚拟连接
        Channel channel = connection.createChannel();

        String input = new Scanner(System.in).next();

        //第一个参数，交换机
        //第二个参数，队列名称，此处不需要
        //第三个参数，额外的设置属性
        //第四个参数，需要发送的消息的字节数组
        channel.basicPublish(RabbitMQConsts.EXCHANGE_WEATHER, "", null, input.getBytes());

        channel.close();
        connection.close();
    }
}
  28. 消费者一代码示例 
    import com.kangswx.rabbitmq.utils.RabbitMQConsts;
import com.kangswx.rabbitmq.utils.RabbitMQUtil;
import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * 百度(模拟消费者)
 */
public class Baidu {

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

        //TCP物理连接
        Connection connection = RabbitMQUtil.getConnection();

        //创建通信通道，相当于TCP的虚拟连接
        Channel channel = connection.createChannel();

        //创建队列，声明并创建一个队列，如果队列已经存在，则使用这个队列
        //第一个参数，对列名称  helloworld
        //第二个参数，是否持久话，false表示不持久化数据，MQ停掉后数据就会丢失
        //第三个参数，是否队列私有化，false表示所有的消费者都可以访问，true表示只有第一次拥有它的消费者才可以一直使用，其他消费者不能访问
        //第四个参数，是否自动删除，false连接停掉后不自动删除掉这个队列
        //第五个参数，其他额外的参数
        channel.queueDeclare(RabbitMQConsts.QUEUE_BAIDU, false, false, false, null);

        //绑定队列交换机
        //第一个参数，队列名称
        //第二个参数，交换机名称
        //第三个参数，路由key，此处不需要
        channel.queueBind(RabbitMQConsts.QUEUE_BAIDU, RabbitMQConsts.EXCHANGE_WEATHER, "");

        //每次只去一条消息进行消费
        channel.basicQos(1);

        channel.basicConsume(RabbitMQConsts.QUEUE_BAIDU, false, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("百度收到气象信息----》"+new String(body));
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        });
    }

}
  29. 消费者二代码示例 
    import com.kangswx.rabbitmq.utils.RabbitMQConsts;
import com.kangswx.rabbitmq.utils.RabbitMQUtil;
import com.rabbitmq.client.*;

import java.io.IOException;

public class Sina {

    public static void main(String[] args) throws IOException {
        //TCP物理连接
        Connection connection = RabbitMQUtil.getConnection();

        //创建通信通道，相当于TCP的虚拟连接
        Channel channel = connection.createChannel();

        //创建队列，声明并创建一个队列，如果队列已经存在，则使用这个队列
        //第一个参数，对列名称  helloworld
        //第二个参数，是否持久话，false表示不持久化数据，MQ停掉后数据就会丢失
        //第三个参数，是否队列私有化，false表示所有的消费者都可以访问，true表示只有第一次拥有它的消费者才可以一直使用，其他消费者不能访问
        //第四个参数，是否自动删除，false连接停掉后不自动删除掉这个队列
        //第五个参数，其他额外的参数
        channel.queueDeclare(RabbitMQConsts.QUEUE_SINA, false, false, false, null);

        //绑定队列交换机
        //第一个参数，队列名称
        //第二个参数，交换机名称
        //第三个参数，路由key，此处不需要
        channel.queueBind(RabbitMQConsts.QUEUE_SINA, RabbitMQConsts.EXCHANGE_WEATHER, "");

        //每次只去一条消息进行消费
        channel.basicQos(1);

        channel.basicConsume(RabbitMQConsts.QUEUE_SINA, false, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("新浪收到气象信息----》"+new String(body));
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        });
    }
}
  30. 路由Routing模式：是在发布订阅模式上的变种
  31. 发布订阅模式是无条件的将所有消息分发给所有的队列，路由模式则是交换机根据Routing key有条件的将数据筛选后分发给消费者队列。
  32. 路由模式下的交换机被称为direct
  33. 创建routing模式下的交换机(类型为direct)
    RabbitMQ介绍及各个工作模式的使用
  34. 路由模式下的生产者 
    import com.kangswx.rabbitmq.utils.RabbitMQConsts;
import com.kangswx.rabbitmq.utils.RabbitMQUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

import java.io.IOException;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
 * 气象局(模拟生产者)
 */
public class WeatherBureau {

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

        LinkedHashMap<String, String> area = new LinkedHashMap<>();
        area.put("china.shaanxi.xian.20190624", "中国陕西西安20190624天气数据");
        area.put("china.shandong.qingdao.20190624", "中国山东青岛20190624天气数据");
        area.put("china.henan.zhengzhou.20190624", "中国河南郑州20190624天气数据");
        area.put("us.cal.la.20190624", "美国加州洛杉矶20190624天气数据");

        area.put("china.shaanxi.xian.20190625", "中国陕西西安20190625天气数据");
        area.put("china.shandong.qingdao.20190625", "中国山东青岛20190625天气数据");
        area.put("china.henan.zhengzhou.20190625", "中国河南郑州20190625天气数据");
        area.put("us.cal.la.20190625", "美国加州洛杉矶20190625天气数据");

        //TCP物理连接
        Connection connection = RabbitMQUtil.getConnection();

        //创建通信通道，相当于TCP的虚拟连接
        Channel channel = connection.createChannel();

        Iterator<Map.Entry<String, String>> iterator = area.entrySet().iterator();
        while(iterator.hasNext()){
            Map.Entry<String, String> next = iterator.next();
            //第一个参数，交换机
            //第二个参数，相当于数据筛选的条件
            //第三个参数，额外的设置属性
            //第四个参数，需要发送的消息的字节数组
            channel.basicPublish(RabbitMQConsts.EXCHANGE_WEATHER_ROUTING, next.getKey(), null, next.getValue().getBytes());
        }

        channel.close();
        connection.close();
    }
}
  35. 路由模式下的消费者 
    import com.kangswx.rabbitmq.utils.RabbitMQConsts;
import com.kangswx.rabbitmq.utils.RabbitMQUtil;
import com.rabbitmq.client.*;

import java.io.IOException;

public class Sina {

    public static void main(String[] args) throws IOException {
        //TCP物理连接
        Connection connection = RabbitMQUtil.getConnection();

        //创建通信通道，相当于TCP的虚拟连接
        Channel channel = connection.createChannel();

        //创建队列，声明并创建一个队列，如果队列已经存在，则使用这个队列
        //第一个参数，对列名称  helloworld
        //第二个参数，是否持久话，false表示不持久化数据，MQ停掉后数据就会丢失
        //第三个参数，是否队列私有化，false表示所有的消费者都可以访问，true表示只有第一次拥有它的消费者才可以一直使用，其他消费者不能访问
        //第四个参数，是否自动删除，false连接停掉后不自动删除掉这个队列
        //第五个参数，其他额外的参数
        channel.queueDeclare(RabbitMQConsts.QUEUE_SINA, false, false, false, null);

        //绑定队列交换机
        //第一个参数，队列名称
        //第二个参数，交换机名称
        //第三个参数，路由key，此处需要设置路由规则
        channel.queueBind(RabbitMQConsts.QUEUE_SINA, RabbitMQConsts.EXCHANGE_WEATHER_ROUTING, "us.cal.la.20190624");
        channel.queueBind(RabbitMQConsts.QUEUE_SINA, RabbitMQConsts.EXCHANGE_WEATHER_ROUTING, "china.henan.zhengzhou.20190624");
        channel.queueBind(RabbitMQConsts.QUEUE_SINA, RabbitMQConsts.EXCHANGE_WEATHER_ROUTING, "us.cal.la.20190625");
        channel.queueBind(RabbitMQConsts.QUEUE_SINA, RabbitMQConsts.EXCHANGE_WEATHER_ROUTING, "china.henan.zhengzhou.20190625");

        //每次只去一条消息进行消费
        channel.basicQos(1);

        channel.basicConsume(RabbitMQConsts.QUEUE_SINA, false, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("新浪收到气象信息----》"+new String(body));
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        });
    }
}
/**
    接收到的消息：
        新浪收到气象信息----》中国河南郑州20190624天气数据
        新浪收到气象信息----》美国加州洛杉矶20190624天气数据
        新浪收到气象信息----》中国河南郑州20190625天气数据
        新浪收到气象信息----》美国加州洛杉矶20190625天气数据
**/
  36. 主题模式Topic：
  37. 在Routing模式的基础上提供了对Routing Key模糊匹配的功能，可以简化程序的编写。
  38. 主题模式下，模糊匹配表达式规则为： 
    * 匹配单个关键字
# 匹配所有关键字
  39. 主题模式下的交换机类型为topic
  40. 创建topic模式下的交换机(类型为topic)：
    RabbitMQ介绍及各个工作模式的使用
  41. 主题模式下的生产者代码示例 
    import com.kangswx.rabbitmq.utils.RabbitMQConsts;
import com.kangswx.rabbitmq.utils.RabbitMQUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

import java.io.IOException;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
 * 气象局(模拟生产者)
 */
public class WeatherBureau {

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

        LinkedHashMap<String, String> area = new LinkedHashMap<>();
        area.put("china.shaanxi.xian.20190624", "中国陕西西安20190624天气数据");
        area.put("china.shandong.qingdao.20190624", "中国山东青岛20190624天气数据");
        area.put("china.henan.zhengzhou.20190624", "中国河南郑州20190624天气数据");
        area.put("us.cal.la.20190624", "美国加州洛杉矶20190624天气数据");

        area.put("china.shaanxi.xian.20190625", "中国陕西西安20190625天气数据");
        area.put("china.shandong.qingdao.20190625", "中国山东青岛20190625天气数据");
        area.put("china.henan.zhengzhou.20190625", "中国河南郑州20190625天气数据");
        area.put("us.cal.la.20190625", "美国加州洛杉矶20190625天气数据");

        //TCP物理连接
        Connection connection = RabbitMQUtil.getConnection();

        //创建通信通道，相当于TCP的虚拟连接
        Channel channel = connection.createChannel();

        Iterator<Map.Entry<String, String>> iterator = area.entrySet().iterator();
        while(iterator.hasNext()){
            Map.Entry<String, String> next = iterator.next();
            //第一个参数，交换机
            //第二个参数，相当于数据筛选的条件
            //第三个参数，额外的设置属性
            //第四个参数，需要发送的消息的字节数组
            channel.basicPublish(RabbitMQConsts.EXCHANGE_WEATHER_TOPIC, next.getKey(), null, next.getValue().getBytes());
        }

        channel.close();
        connection.close();
    }
}
  42. 主题模式的消费者示例1 
    import com.kangswx.rabbitmq.utils.RabbitMQConsts;
import com.kangswx.rabbitmq.utils.RabbitMQUtil;
import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * 百度(模拟消费者)
 */
public class Baidu {

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

        //TCP物理连接
        Connection connection = RabbitMQUtil.getConnection();

        //创建通信通道，相当于TCP的虚拟连接
        Channel channel = connection.createChannel();

        //创建队列，声明并创建一个队列，如果队列已经存在，则使用这个队列
        //第一个参数，对列名称  helloworld
        //第二个参数，是否持久话，false表示不持久化数据，MQ停掉后数据就会丢失
        //第三个参数，是否队列私有化，false表示所有的消费者都可以访问，true表示只有第一次拥有它的消费者才可以一直使用，其他消费者不能访问
        //第四个参数，是否自动删除，false连接停掉后不自动删除掉这个队列
        //第五个参数，其他额外的参数
        channel.queueDeclare(RabbitMQConsts.QUEUE_BAIDU, false, false, false, null);

        //绑定队列交换机
        //第一个参数，队列名称
        //第二个参数，交换机名称
        //第三个参数，路由key，此处需要设置路由规则
        channel.queueBind(RabbitMQConsts.QUEUE_BAIDU, RabbitMQConsts.EXCHANGE_WEATHER_TOPIC, "us.#");

        //每次只去一条消息进行消费
        channel.basicQos(1);

        channel.basicConsume(RabbitMQConsts.QUEUE_BAIDU, false, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("百度收到气象信息----》"+new String(body));
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        });
    }

}
  43. 主题模式的消费者示例2 
    import com.kangswx.rabbitmq.utils.RabbitMQConsts;
import com.kangswx.rabbitmq.utils.RabbitMQUtil;
import com.rabbitmq.client.*;

import java.io.IOException;

public class Sina {

    public static void main(String[] args) throws IOException {
        //TCP物理连接
        Connection connection = RabbitMQUtil.getConnection();

        //创建通信通道，相当于TCP的虚拟连接
        Channel channel = connection.createChannel();

        //创建队列，声明并创建一个队列，如果队列已经存在，则使用这个队列
        //第一个参数，对列名称  helloworld
        //第二个参数，是否持久话，false表示不持久化数据，MQ停掉后数据就会丢失
        //第三个参数，是否队列私有化，false表示所有的消费者都可以访问，true表示只有第一次拥有它的消费者才可以一直使用，其他消费者不能访问
        //第四个参数，是否自动删除，false连接停掉后不自动删除掉这个队列
        //第五个参数，其他额外的参数
        channel.queueDeclare(RabbitMQConsts.QUEUE_SINA, false, false, false, null);

        //绑定队列交换机
        //第一个参数，队列名称
        //第二个参数，交换机名称
        //第三个参数，路由key，此处需要设置路由规则
        channel.queueBind(RabbitMQConsts.QUEUE_SINA, RabbitMQConsts.EXCHANGE_WEATHER_TOPIC, "*.*.*.20190624");

        //解绑
        //channel.queueUnbind(RabbitMQConsts.QUEUE_SINA, RabbitMQConsts.EXCHANGE_WEATHER_TOPIC, "*.*.*.20190624");

        //每次只去一条消息进行消费
        channel.basicQos(1);

        channel.basicConsume(RabbitMQConsts.QUEUE_SINA, false, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("新浪收到气象信息----》"+new String(body));
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        });
    }
}
  44. RabbitMQ消息确认机制：RabbitMQ在传递消息的过程中充当了代理人(Broker)的角色，生产者怎么知道消息被正确投递到了Broker了呢？
  45. RabbitMQ提供了监听器(Listener)来接收消息的投递状态。
  46. 消息确认涉及两种状态Confirm与Return。
  47. Confirm代表生产者将消息送到Broker时产生的状态，后续会出现两种情况：
    1. ack 代表broker已经将数据接收
    2. nack 代表broker拒收消息，原因有多种，队列已满、限流、IO异常…
  48. Return代表消息被Broker正常接收(ack)后，但Broker没有对应的队列进行投递时产生的状态，消息被退回生产者。
  49. 上面的两种状态只代表生产者与Broker之间消息投递的情况。与消费者是否接收/确认消息无关。
  50. 监听器的代码示例 
    import com.kangswx.rabbitmq.utils.RabbitMQConsts;
import com.kangswx.rabbitmq.utils.RabbitMQUtil;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
 * 气象局(模拟生产者)
 */
public class WeatherBureau {

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

        LinkedHashMap<String, String> area = new LinkedHashMap<>();
        area.put("china.shaanxi.xian.20190624", "中国陕西西安20190624天气数据");
        area.put("china.shandong.qingdao.20190624", "中国山东青岛20190624天气数据");
        area.put("china.henan.zhengzhou.20190624", "中国河南郑州20190624天气数据");
        area.put("us.cal.la.20190624", "美国加州洛杉矶20190624天气数据");

        area.put("china.shaanxi.xian.20190625", "中国陕西西安20190625天气数据");
        area.put("china.shandong.qingdao.20190625", "中国山东青岛20190625天气数据");
        area.put("china.henan.zhengzhou.20190625", "中国河南郑州20190625天气数据");
        area.put("us.cal.la.20190625", "美国加州洛杉矶20190625天气数据");

        //TCP物理连接
        Connection connection = RabbitMQUtil.getConnection();

        //创建通信通道，相当于TCP的虚拟连接
        Channel channel = connection.createChannel();

        //开启confirm监听模式
        channel.confirmSelect();
        //添加监听器的处理代码
        channel.addConfirmListener(new ConfirmListener() {
            //成功接收时处理的代码
            //第一个参数是消息在传递的过程中的唯一id，第二个参数是数据是否为批量接收(一般用不到)
            @Override
            public void handleAck(long l, boolean b) throws IOException {
                System.out.println("消息成功投递，Tag:"+l);
            }

            //MQ拒收时处理的代码
            @Override
            public void handleNack(long l, boolean b) throws IOException {
                System.out.println("消息投递被拒收，Tag:"+l);
            }
        });

        //添加ReturnListener监听器
        /*channel.addReturnListener(new ReturnListener() {
            @Override
            public void handleReturn(int i, String s, String s1, String s2, AMQP.BasicProperties basicProperties, byte[] bytes) throws IOException {

            }
        });*/
        channel.addReturnListener(new ReturnCallback() {
            @Override
            public void handle(Return r) {
                System.err.println("=================");
                System.err.println("Return编码："+r.getReplyCode()+",描述信息："+r.getReplyText());
                System.err.println("交换机："+r.getExchange()+",路由key："+r.getRoutingKey());
                System.err.println("消息主题为："+new String(r.getBody()));
                System.err.println("=================");
            }
        });

        Iterator<Map.Entry<String, String>> iterator = area.entrySet().iterator();
        while(iterator.hasNext()){
            Map.Entry<String, String> next = iterator.next();
            //第一个参数，交换机
            //第二个参数，相当于数据筛选的条件
            //第三个参数，mandatory，为true时如果消息无法正常投递到交换机则return回生产者，为false时直接将消息丢弃
            //第四个参数，额外的设置属性
            //第五个参数，需要发送的消息的字节数组
            channel.basicPublish(RabbitMQConsts.EXCHANGE_WEATHER_TOPIC, next.getKey(), true, null, next.getValue().getBytes());
        }

        //让channel一直处于等待监听的状态
        /*channel.close();
        connection.close();*/
    }
}
  51. Exchange模式是生产者和消费者之间通信需要通过交换机(Exchange)绑定队列(Queue)进行分发。
  52. Exchange对应fanout，direct，topic三种类型，
    RabbitMQ介绍及各个工作模式的使用
  53. 上面所有的代码可见RabbitMQ使用示例代码
rabbitmq rabbitmq工作模式 RabbitMQ投递消息监听
上一篇: 纯Html+Ajax和JSP两者对比的个人理解
下一篇: html和jsp的区别--静态页面和动态页面的区别

继续阅读