RabbitMQ消息中间件_大数据系统

工作原理

Broker:接收和分发消息的应用，RabbitMQ Server就是Message Broker
**Virtual host:**出于多租户和安全因素设计的，把 AMQP 的基本组件划分到一个虚拟的分组中，类似
于网络中的namespace概念。当多个不同的用户使用同一个RabbitMQ server提供的服务时，可以划分出
多个vhost，每个用户在自己的vhost创建exchange / queue等
Connection: publisher / consumer和broker之间的TCP连接
Channel:如果每一次访问RabbitMQ都建立一个Connection，在消息量大的时候建立TCP
Connection的开销将是巨大的，效率也较低。Channel是在connection内部建立的逻辑连接，如果应用程
序支持多线程，通常每个thread创建单独的channel进行通讯，AMQPmethod包含了channel id帮助客
户端和message broker识别channel，所以channel之间是完全隔离的。Channel作为轻量级的
Connection极大减少了操作系统建立TCP connection的开销

什么是MQ： MQ全称 Message Queue（消息队列），是在消息的传输过程中保存消息的容器。多用于分布式系统之间进行通信

1 优点:

1.应用解耦: 降低系统之间的耦合，提高系统的可维护性。
2.异步提速: 可以系统的吞吐量。
3.削峰填谷: 可以提高系统的稳定性。

2 缺点：

系统可用性降低
系统引入的外部依赖越多，系统稳定性越差。一旦 MQ 宕机，就会对业务造成影响。如何保证MQ的高可用？

系统复杂度提高
MQ 的加入大大增加了系统的复杂度，以前系统间是同步的远程调用，现在是通过 MQ 进行异步调用。如何保证消息没有被重复消费？怎么处理消息丢失情况？那么保证消息传递的顺序性？

一致性问题
A 系统处理完业务，通过 MQ 给B、C、D三个系统发消息数据，如果 B 系统、C 系统处理成功，D 系统处理失败。如何保证消息数据处理的一致性？

rabbitmq的工作模式:

1. 简单模式
2. worker模式
3. 分发模式
4. 路由模式
5. 主题模式

1. 简单模式

所需要的依赖


        
        
            com.rabbitmq
            amqp-client
            5.12.0

生产者

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

public class Product {
    public static void main(String[] args) throws Exception {
         //1. 创建连接对象工厂类
        ConnectionFactory factory=new ConnectionFactory();
        //2. 设置工厂对象的配置。
        factory.setHost("1虚拟机的IP");
        factory.setPort(5672); //AMQP: 5672  Http:15672 集群:25672
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");

        //3.获取连接对象
        Connection connection=factory.newConnection();

        //4.获取通道Channel
        Channel channel = connection.createChannel();

        //5. 创建队列 如果队列名存在 则不创建  如果不存在则创建新的队列。
        
        channel.queueDeclare("simple_queue",true,false,false,null);

        //6. 发送消息。
        
        String body="hello world~~~~~~~~~1";
        channel.basicPublish("","simple_queue",null,body.getBytes());
        //7. 关闭资源
        connection.close();
    }
}

消费者


import com.rabbitmq.client.*;

import java.io.IOException;


public class Consumer01 {
    public static void main(String[] args) throws Exception {
        //1. 创建连接对象工厂类
        ConnectionFactory factory=new ConnectionFactory();
        //2. 设置工厂对象的配置。
        factory.setHost("虚拟机IP");
        factory.setPort(5672); //AMQP: 5672  Http:15672 集群:25672
        factory.setVirtualHost("/");
        factory.setUsername("guest");
        factory.setPassword("guest");
        //3.获取连接对象
        Connection connection=factory.newConnection();
        //4.获取通道Channel
        Channel channel = connection.createChannel();
        //5. 如果队列存在则使用原来的 否则创建新的
        channel.queueDeclare("simple_queue",true,false,false,null);

        //6. 消费队列中的消息。
        
        Consumer consumer=new DefaultConsumer(channel){
            
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("consumerTag:"+consumerTag);
                System.out.println("RoutingKey:"+envelope.getRoutingKey());
                System.out.println("Exchange:"+envelope.getExchange());
                System.out.println("DeliveryTag:"+envelope.getDeliveryTag());
                System.out.println("消息的内容:"+new String(body));
            }
        };
        channel.basicConsume("simple_queue",true,consumer);

        //要不要关闭消费者连接? 
        //一定不要关闭。关闭就不能接受信息
    }
}

work模式

生产者


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

public class Product {
    public static void main(String[] args)throws  Exception {
        //创建连接工厂 --配置连接信息
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost("虚拟机IP");//log 192.168.235.136
        //创建连接对象Connection
        Connection connection=factory.newConnection();
        //创建信道
        Channel channel = connection.createChannel();

        //创建队列
        
        channel.queueDeclare("queue_work",true,false,false,null);
        //发生消息
        
        for(int i=0;i<10;i++) {
            String msg = "love"+i;
            channel.basicPublish("", "queue_work", null, msg.getBytes());
        }
        //生产者这里可以管理资源  消费者不能关闭资源。
        channel.close();
        connection.close();

    }
}

消费者

与之前的消费者一致.

发布订阅模式

引入了交换机

生产者

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


public class Product {
    public static void main(String[] args)throws  Exception {
        //创建连接工厂 --配置连接信息
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost("192.168.213.188");
        //创建连接对象Connection
        Connection connection=factory.newConnection();
        //创建信道
        Channel channel = connection.createChannel();

        //创建队列
        
        channel.queueDeclare("queue_fanout01",true,false,false,null);
        channel.queueDeclare("queue_fanout02",true,false,false,null);

        //创建交换机
        
        channel.exchangeDeclare("exchange", BuiltinExchangeType.FANOUT,true);
        
        channel.queueBind("queue_fanout01","exchange","");
        channel.queueBind("queue_fanout02","exchange","");
        //发生消息
        
        for(int i=0;i<10;i++) {
            String msg = "love"+i;
            channel.basicPublish("exchange", "", null, msg.getBytes());
        }
        //生产者这里可以管理资源  消费者不能关闭资源。
        channel.close();
        connection.close();

    }
}

路由模式

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



public class Product {
    public static void main(String[] args)throws  Exception {
        //创建连接工厂 --配置连接信息
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost("192.168.213.188");
        //创建连接对象Connection
        Connection connection=factory.newConnection();
        //创建信道
        Channel channel = connection.createChannel();

        //创建队列
        
        channel.queueDeclare("queue_direct01",true,false,false,null);
        channel.queueDeclare("queue_direct02",true,false,false,null);

        //创建交换机
        
        channel.exchangeDeclare("exchange_direct", BuiltinExchangeType.DIRECT,true);
        
        channel.queueBind("queue_direct01","exchange_direct","error");


        channel.queueBind("queue_direct02","exchange_direct","info");
        channel.queueBind("queue_direct02","exchange_direct","error");
        channel.queueBind("queue_direct02","exchange_direct","warning");
        //发生消息
        
        for(int i=0;i<10;i++) {
            String msg = "love！"+i;
            channel.basicPublish("exchange_direct", "error", null, msg.getBytes());
        }
        //生产者这里可以管理资源  消费者不能关闭资源。
        channel.close();
        connection.close();

    }
}

topic主体模式消费者

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


public class Product {
    public static void main(String[] args)throws  Exception {
        //创建连接工厂 --配置连接信息
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost("192.168.213.188");
        //创建连接对象Connection
        Connection connection=factory.newConnection();
        //创建信道
        Channel channel = connection.createChannel();

        //创建队列
        
        channel.queueDeclare("queue_topic01",true,false,false,null);
        channel.queueDeclare("queue_topic02",true,false,false,null);

        //创建交换机
        
        channel.exchangeDeclare("exchange_topic", BuiltinExchangeType.TOPIC,true);
        
        channel.queueBind("queue_topic01","exchange_topic","*.orange.*");


        channel.queueBind("queue_topic02","exchange_topic","*.*.rabbit");
        channel.queueBind("ueue_topic02","xchange_topic","lazy.#");


        //发生消息
        
        for(int i=0;i<10;i++) {
            String msg = "love"+i;
            channel.basicPublish("exchange_topic", "lazy.orange.rabbit", null, msg.getBytes());
        }
        //生产者这里可以管理资源  消费者不能关闭资源。
        channel.close();
        connection.close();

    }
}

回顾

枚举类型：

public enum Color {  
2   RED, GREEN, BLANK, YELLOW  
3 }

Jackson ObjectMapper类(com.fasterxml.jackson.databind.ObjectMapper)
是使用Jackson解析JSON最简单的方法。Jackson ObjectMapper可以从字符串、流或文件解析JSON，并创建Java对象或对象图来表示已解析的JSON。将JSON解析为Java对象也称为从JSON反序列化Java对象

Linux开启rabbit服务

systemctl start rabbitmq-server

RabbitMQ消息中间件

大数据系统相关栏目本月热门文章