[大数据]rabbitmq

rabbitmq

1、AMQP

AMQP，即Advanced Message Queuing Protocol,一个提供统一消息服务的应用层标准高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息，并不受客户端/中间件不同产品，不同的开发语言等条件的限制。

2、RabbitMQ的使用

1、核心概念

Message

消息，消息是不具名的，它由消息头和消息体组成。消息体是不透明的，而消息头则由一系列的可选属性组成，这些属性包括routing-key（路由键）、 priority（相对于其他消息的优先权）、 delivery-mode（指出该消息可能需要持久性存储）等。

Publisher

消息的生产者，也是一个向交换器发布消息的客户端应用程序

Exchange

交换器，用来接收生产者发送的消息并将这些消息路由给服务器中的队列。Exchange有4种类型： direct(默认)， fanout, topic, 和headers，不同类型的Exchange转发消息的策略有所区别

Queue

消息队列，用来保存消息直到发送给消费者。它是消息的容器，也是消息的终点。一个消息可投入一个或多个队列。消息一直在队列里面，等待消费者连接到这个队列将其取走。

Binding

绑定，用于消息队列和交换器之间的关联。一个绑定就是基于路由键将交换器和消息队列连
接起来的路由规则，所以可以将交换器理解成一个由绑定构成的路由表。Exchange 和Queue的绑定可以是多对多的关系。
Connection：网络连接，比如一个TCP连接。
Channel：信道，多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的TCP连接内的虚拟连接， AMQP 命令都是通过信道发出去的，不管是发布消息、订阅队列还是接收消息，这些动作都是通过信道完成。因为对于操作系统来说建立和销毁 TCP 都是非常昂贵的开销，所以引入了信道的概念，以复用一条 TCP 连接。
Consumer：消息的消费者，表示一个从消息队列中取得消息的客户端应用程序。
Virtual Host：虚拟主机，表示一批交换器、消息队列和相关对象。虚拟主机是共享相同的身份认证和加密环境的独立服务器域。每个 vhost 本质上就是一个 mini 版的 RabbitMQ 服务器，拥有自己的队列、交换器、绑定和权限机制。 vhost 是 AMQP 概念的基础，必须在连接时指定，RabbitMQ 默认的 vhost 是 / 。

2、几种消息模型

• 简单模式

概述：一个生产者一个消费者

应用场景：一发一接的结构

例如: 手机短信,邮件单发

//发送者
private static String QUEUE_NAME = "test-simple";

    //创建队列发送消息
    public  void sendMessage() throws Exception{
        //获取连接
        Connection connection = getConnection();
        //创建通道
        Channel channel = connection.createChannel();
        //创建队列
        //队列名称 是否持久化 是否独占队列 是否消费完成后删除队列 额外附加参数
        channel.queueDeclare(QUEUE_NAME,false,false,false,null);
        //消息内容
        String message = "hello tom!";
        //交换机名称 队列名称 传递消息额外设置 消息的具体内容
        channel.basicPublish("",QUEUE_NAME,null,message.getBytes(StandardCharsets.UTF_8));
        System.out.println("发送消息："+message);
        //关闭通道、连接
        channel.close();
        connection.close();
    }
//接收者
 private static String QUEUE_NAME = "test-simple";

    public void getMessage() throws Exception {
        //获取连接
        Connection connection = getConnection();
        //创建通道
        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        channel.basicConsume(QUEUE_NAME, true, 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.close();
        connection.close();
    }

• 工作模式(资源争抢)

概述：一个生产者多个消费者

应用场景：一发多收结构

例如：抢红包、资源调度

//生产者
private static String QUEUE_NAME = "test-work";

    public static void main(String[] args) throws Exception {
        //获取连接
        Connection connection = getConnection();
        //创建通道
        Channel channel = connection.createChannel();
        //创建队列
        //队列名称 是否持久化 是否独占队列 是否消费完成后删除队列 额外附加参数
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        //消息内容
        for (int i = 0; i < 20; i++) {
            //交换机名称 队列名称 传递消息额外设置 消息的具体内容
            channel.basicPublish("", QUEUE_NAME, null, ("hello tom!"+i).getBytes(StandardCharsets.UTF_8));
            System.out.println("发送+"+i);
        }
        System.out.println("发送消息: success");
        //关闭通道、连接
        channel.close();
        connection.close();
    }

//消费者
 private static String QUEUE_NAME = "test-work";

    public static void main(String[] args) throws Exception {
        //获取连接
        Connection connection = getConnection();
        //创建通道
        Channel channel = connection.createChannel();

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

        channel.basicConsume(QUEUE_NAME, true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者1-收到消息：" + new String(body));
                channel.basicAck(1,false);
            }
        });
    }

• 发布订阅模式

概述：发送消息到交换机 交换机根据绑定队列发送给生产者

应用场景：服务推送

例如：订阅公众号接收文档

//发布者
  private static String EXCHANGE_NAME = "fanout-exchange";
    
    public static void main(String[] args) throws Exception {
        //获取连接
        Connection connection = getConnection();
        Channel channel = connection.createChannel();
        //交换机名称 交换机类型 fanout广播类型
        channel.exchangeDeclare(EXCHANGE_NAME,"fanout");
        channel.basicPublish(EXCHANGE_NAME,"",null,"fanout type message".getBytes(StandardCharsets.UTF_8));
        //
        channel.close();
        connection.close();
    }
//订阅者
private static String EXCHANGE_NAME = "fanout-exchange";

    public static void main(String[] args) throws Exception {
        //获取连接
        Connection connection = getConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(EXCHANGE_NAME,"fanout");
        //临时队列
        String queue = channel.queueDeclare().getQueue();
        //绑定交换机和队列
        channel.queueBind(queue,EXCHANGE_NAME,"");
        //消费消息
        channel.basicConsume(queue,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者2："+new String(body));
            }
        });
    }

• 路由模式

概述：根据routKey不同，交换机发送到不同队列中。

//生产者

    private static String DIRECT_NAME = "direct_exchange";

    public static void main(String[] args) throws Exception {
        //获取连接
        Connection connection = getConnection();
        Channel channel = connection.createChannel();
        //交换机名称 交换机类型 direct路由模式
        channel.exchangeDeclare(DIRECT_NAME,"direct");
        //发送消息
        String routingKey = "warning";
        channel.basicPublish(DIRECT_NAME,routingKey,null,("这是direct模式发布routKey:["+routingKey+"],发送新增消息").getBytes(StandardCharsets.UTF_8));
        channel.close();
        connection.close();
    }
//消费者
private static String DIRECT_NAME = "direct_exchange";

    public static void main(String[] args) throws Exception {
        //获取连接
        Connection connection = getConnection();
        Channel channel = connection.createChannel();
        //交换机名称 交换机类型 direct路由模式
        channel.exchangeDeclare(DIRECT_NAME,"direct");
        String queue = channel.queueDeclare().getQueue();
        //临时队列和交换机绑定
        channel.queueBind(queue,DIRECT_NAME,"info");
        channel.queueBind(queue,DIRECT_NAME,"error");
        channel.queueBind(queue,DIRECT_NAME,"warning");
        //消费消息
        channel.basicConsume(queue,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者2："+new String(body));
            }
        });
    }

• 统配符匹配模式

//生产者
  private static String TOPIC_NAME = "topic_exchange";

    public static void main(String[] args) throws Exception {
        //获取连接
        Connection connection = getConnection();
        Channel channel = connection.createChannel();
        //交换机名称 交换机类型 topic
        channel.exchangeDeclare(TOPIC_NAME,"topic");
        //发布消息
        String routingKey = "user.service";
        channel.basicPublish(TOPIC_NAME,routingKey,null,("这是topic动态路由模型,routKey["+routingKey).getBytes(StandardCharsets.UTF_8));
        channel.close();
        connection.close();
    }
//消费者
 private static String TOPIC_NAME = "topic_exchange";

    public static void main(String[] args) throws Exception {
        //获取连接
        Connection connection = getConnection();
        Channel channel = connection.createChannel();
        //交换机名称 交换机类型 topic
        channel.exchangeDeclare(TOPIC_NAME,"topic");
        //创建一个临时队列
        String queue = channel.queueDeclare().getQueue();
        //绑定队列和交换机 动态通配符形式
        channel.queueBind(queue,TOPIC_NAME,"user.*");
        //消费消息
        channel.basicConsume(queue,true,new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者2："+new String(body));
            }
        });
    }

springboot与rabbitmq整合

• 导入依赖

 <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
 </dependency>

• 配置连接信息

spring:
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: test
    password: 123456
    virtual-host: /test

生产者

 @Autowired
    private RabbitTemplate rabbitTemplate;
    @Test
    void testHello() {
        for (int i = 0; i < 10; i++) {
            rabbitTemplate.convertAndSend("work","hello world："+i);
        }

    }

消费者

@Component
public class HelloConsumer {

    @RabbitListener(queuesToDeclare = @Queue("work"))
    public void receive(String message){
        System.out.println("message1："+message);
    }

    @RabbitListener(queuesToDeclare = @Queue("work"))
    public void receive2(String message){
        System.out.println("message2："+message);
    }
}

• 发布订阅模式

发送端

    //发布订阅模式
    @Test
    public void testFanout(){
        rabbitTemplate.convertAndSend("logs","","Fanout的模型发送消息");
    }

接收端

@Component
public class FanoutConsumer {
    @RabbitListener(bindings = {
            @QueueBinding(
                    value = @Queue,//创建临时队列
                    exchange = @Exchange(value = "logs",type = "fanout") //绑定的交换机
            )
    })
    public void receive(String message){
        System.out.println("message1 = " +message);
    }

    @RabbitListener(bindings = {
            @QueueBinding(
                    value = @Queue,//创建临时队列
                    exchange = @Exchange(value = "logs",type = "fanout") //绑定的交换机
            )
    })
    public void receive2(String message){
        System.out.println("message2 = " +message);
    }
}

• 路由模式

发送端

 //路由模式
    @Test
    public void testDirect(){
        rabbitTemplate.convertAndSend("directs","error","Direct的模式发送信息");
    }

接收端

@Component
public class DirectConsumer {

    @RabbitListener(bindings = {
            @QueueBinding(
                    value = @Queue,//创建临时队列
                    exchange = @Exchange(value = "directs", type = "direct"),//自定义交换机名称和类型
                    key = {"info", "error", "warn"}
            )
    })
    public void receive1(String message) {
        System.out.println("message1 = " + message);
    }

    @RabbitListener(bindings = {
            @QueueBinding(
                    value = @Queue,//创建临时队列
                    exchange = @Exchange(value = "directs", type = "direct"),//自定义交换机名称和类型
                    key = {"error"}
            )
    })
    public void receive2(String message) {
        System.out.println("message2 = " + message);
    }
}

• 统配符模式

发送端

    //统配符模式
    @Test
    void testTopic(){
        rabbitTemplate.convertAndSend("topics","user.ll","这是topic模式发送信息");
    }

接收端

@Component
public class TopicConsumer {

    @RabbitListener(bindings = {
            @QueueBinding(
                    value = @Queue,//临时队列
                    exchange = @Exchange(name = "topics",type = "topic"),
                    key = {"user.save","user.*"}
            )
    })
    public void receive1(String message){
        System.out.println("message1 = "+message);
    }

    @RabbitListener(bindings = {
            @QueueBinding(
                    value = @Queue,//临时队列
                    exchange = @Exchange(name = "topics",type = "topic"),
                    key = {"user.get"}
            )
    })
    public void receive2(String message){
        System.out.println("message2 = "+message);
    }
}

  )
})
public void receive1(String message){
    System.out.println("message1 = "+message);
}

@RabbitListener(bindings = {
        @QueueBinding(
                value = @Queue,//临时队列
                exchange = @Exchange(name = "topics",type = "topic"),
                key = {"user.get"}
        )
})
public void receive2(String message){
    System.out.println("message2 = "+message);
}

}

---