RabbitMQ的基础应用

MQ概念：

MQ全称为Message Queue (消息队列) 又称 (消息中间件)，是在消息的传输过程中保存消息的容器，多用于分布式系统之间的通信。通信方式：第一种：通过webservice直接远程调用。第二种：通过中间件操作，zookeeper其实就是中间件。

MQ的优点:

1.可以解耦合，如果此时每增加一个消费者，都需要通过修改生产者源码来增加接口，此时耦合非常高，但是如果使用中间件的话，生产者只需要发送一次消息到中间件，别的消费者就能复用该信息，当增加或删除消费者调用接口的时候，不需要额外的更新代码，可维护性提高。当其中一个消费者出问题时也不会牵连到其他消费者，容错性变高。

2.可以异步提速，生产者和消费者是一个整体，每次操作都要一起操作花费的时间较长，当使用中间件后，生产者和消费者的操作可以分开，大大提高了操作所花费的时间，就像订单系统那样

3.削峰填谷，假如有个A系统它每秒可以处理10000次请求，当有50000次请求时，系统就会瞬间崩溃，当有了中间件后可以把50000次请求存到中间件里，然后每次从中间件里取出10000次请求交给A系统处理，这样系统就不会崩溃了。当然也会产生一些问题 :1.对MQ的压力比较大 2.会让就我们的系统长时间的处于最高性能消耗

MQ的缺点:

1.系统可用性降低

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

2.系统复杂度提高

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

3.一致性问题

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

在什么情况下可以考虑使用MQ

①生产者不需要从消费者处获得反馈。引入消息队列之前的直接调用，其接口的返回值应该为空，这才让明明下层的动作还没做，上层却当成动作做完了继续往后走，即所谓异步成为了可能。

②容许短暂的不一致性。

③确实是用了有效果。即解耦、提速、削峰这些方面的收益，超过加入MQ，管理MQ这些成本。

常见的MQ产品：RabbitMQ ，ActiveMQ，RocketMQ，Kafka

RabbitMQ 是基于 AMQP 协议使用 Erlang 语言开发的一款消息队列产品。AMQP：高级消息队列协议

RabbitMQ简易模式：

1.创建两个工程，生产者和消费者。2.为两个工程添加依赖

 <dependencies><!--rabbitmq java 客户端--><dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.6.0</version></dependency></dependencies><build><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><version>3.8.0</version><configuration><source>1.8</source><target>1.8</target></configuration></plugin></plugins></build>

3.配置生产者

public class Producertest {public static void main(String[] args) throws IOException, TimeoutException {//创建连接工厂ConnectionFactory factory = new ConnectionFactory ();//设置参数factory.setHost ( "localhost" );//ip,默认值为localhostfactory.setPort ( 5672 );//端口，默认为5672factory.setVirtualHost ( "/itcst" );//虚拟机默认值/factory.setUsername ( "huangshu" );//用户名，默认值为guestfactory.setPassword ( "huangshu" );//密码，默认值为guest//创建连接 ConnectionConnection connection = factory.newConnection ();//创建ChannelChannel channel = connection.createChannel ();//创建消息队列Queue/*queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)参数：1. queue：队列名称2. durable:是否持久化，当mq重启之后，还在3. exclusive：* 是否独占。只能有一个消费者监听这队列* 当Connection关闭时，是否删除队列*4. autoDelete:是否自动删除。当没有Consumer时，自动删除掉5. arguments：参数。*/channel.queueDeclare ( "hehe" , true , false , false , null );//发送消息String body = "我是一个程序员";/*basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)参数：1. exchange：交换机名称。简单模式下交换机会使用默认的 ""2. routingKey：路由名称3. props：配置信息4. body：发送消息数据*/channel.basicPublish ( "" , "hehe" , null , body.getBytes () );//释放资源channel.close ();connection.close ();}
}

配置消费者

public class Cousumer {public static void main(String[] args) throws IOException, TimeoutException {//创建连接工厂ConnectionFactory factory = new ConnectionFactory ();//设置参数factory.setHost ( "localhost" );//ip,默认值为localhostfactory.setPort ( 5672 );//端口，默认为5672factory.setVirtualHost ( "/itcst" );//虚拟机默认值/factory.setUsername ( "huangshu" );//用户名，默认值为guestfactory.setPassword ( "huangshu" );//密码，默认值为guest//创建连接 ConnectionConnection connection = factory.newConnection ();//创建ChannelChannel channel = connection.createChannel ();//创建队列Queue/*queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)参数：1. queue：队列名称2. durable:是否持久化，当mq重启之后，还在3. exclusive：* 是否独占。只能有一个消费者监听这队列* 当Connection关闭时，是否删除队列*4. autoDelete:是否自动删除。当没有Consumer时，自动删除掉5. arguments：参数。*/channel.queueDeclare ( "hehe" , true , false , false , null );//接受消息/*basicConsume(String queue,boolean autoAck, Consumer callback)参数：1.queue：队列名称2.autoAck：是否自动确认3.Consumer：回调对象*/Consumer consumer = new DefaultConsumer ( channel ) {/***监听消息，当有消息的时候，会调用此方法接受消息，消息会存放方法的参数中，自动执行1. consumerTag：标识2. envelope：获取一些信息，交换机，路由key...3. properties:配置信息4. body：数据*/public void handleDelivery(String consumerTag , Envelope envelope , AMQP.BasicProperties properties , byte[] body) throws IOException {System.out.println("consumerTag："+consumerTag);System.out.println("Exchange："+envelope.getExchange());System.out.println("RoutingKey："+envelope.getRoutingKey());System.out.println("properties："+properties);System.out.println("body："+new String(body));}};channel.basicConsume ( "hehe" , true , consumer );}
}

RabbitMQ工作队列模式：与简易模式相比多了一个或一些消费者，

1.代码与简易模式相似，只不过有多个消费者

2.多个消费端消费同一个队列的消息，对于同一个消息的关系是竞争关系

RabbitMQ发布订阅模式:

特点：生产者发送消息，这个消息可以被多个消费者消费

过程：生产者将消息发送给交换机，交换机通过路由操作，将消息发送给不同队列，这样消费者就可以从不同队列中去取出消息

Exchange有以下三种常见类型：

Fanout:广播，将消息发送给所有绑定交换机的队列

Direck:定向，把消息交给指定routingKey（路由规则）的队列

Topic:通配符，把消息交给指定routing pattern（路由模式）的队列

生产者:

public class Producer_PubSub {public static void main(String[] args) throws IOException, TimeoutException {//创建连接工厂ConnectionFactory factory = new ConnectionFactory ();//设置参数factory.setHost ( "localhost" );factory.setPort ( 5672 );factory.setUsername ( "huangshu" );factory.setPassword ( "huangshu" );//创建连接 ConnectionConnection connection = factory.newConnection ();//创建ChannelChannel channel = connection.createChannel ();//创建交换机/*exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)参数：1. exchange:交换机名称2. type:交换机类型DIRECT("direct"),：定向FANOUT("fanout"),：扇形（广播），发送消息到每一个与之绑定队列。TOPIC("topic"),通配符的方式HEADERS("headers");参数匹配3. durable:是否持久化4. autoDelete:自动删除5. internal：内部使用。 一般false6. arguments：参数*/String exchangeName = "test_fanout";//交换机名称channel.exchangeDeclare ( exchangeName , BuiltinExchangeType.FANOUT , true , false , false , null );//创建队列String queue1Name = "test_fanout_queue1";//队列名称String queue2Name = "test_fanout_queue2";//队列名称channel.queueDeclare ( queue1Name , true , false , false , null );channel.queueDeclare ( queue2Name , true , false , false , null );//绑定交换机和队列//7. 绑定队列和交换机/*queueBind(String queue, String exchange, String routingKey)参数：1. queue：队列名称2. exchange：交换机名称3. routingKey：路由键，绑定规则如果交换机的类型为fanout ，routingKey设置为""*/channel.queueBind ( queue1Name , exchangeName , "" );channel.queueBind ( queue2Name , exchangeName , "" );//发送消息String body = "日志信息:我来自斗气化马恐怖如斯，信息等级：info";channel.basicPublish ( exchangeName , "" , null , body.getBytes () );//释放资源channel.close ();connection.close ();}}

消费者有多个，这里我只写了一个：

public class Cousumer_PubSub1 {public static void main(String[] args) throws IOException, TimeoutException {//创建连接工厂ConnectionFactory factory = new ConnectionFactory ();//设置参数factory.setHost ( "localhost" );//ip,默认值为localhostfactory.setPort ( 5672 );//端口，默认为5672factory.setVirtualHost ( "/itcst" );//虚拟机默认值/factory.setUsername ( "huangshu" );//用户名，默认值为guestfactory.setPassword ( "huangshu" );//密码，默认值为guest//创建连接 ConnectionConnection connection = factory.newConnection ();//创建ChannelChannel channel = connection.createChannel ();//创建队列QueueString queue1Name = "test_fanout_queue1";//队列名称String queue2Name = "test_fanout_queue2";//队列名称//接受消息/*basicConsume(String queue,boolean autoAck, Consumer callback)参数：1.queue：队列名称2.autoAck：是否自动确认3.Consumer：回调对象*/Consumer consumer = new DefaultConsumer ( channel ) {/***监听消息，当有消息的时候，会调用此方法接受消息，消息会存放方法的参数中，自动执行1. consumerTag：标识2. envelope：获取一些信息，交换机，路由key...3. properties:配置信息4. body：数据*/public void handleDelivery(String consumerTag , Envelope envelope , AMQP.BasicProperties properties , byte[] body) throws IOException {//                System.out.println("consumerTag："+consumerTag);
//                System.out.println("Exchange："+envelope.getExchange());
//                System.out.println("RoutingKey："+envelope.getRoutingKey());
//                System.out.println("properties："+properties);System.out.println("body："+new String(body));System.out.println ("将日志信息打印到控制台......");}};channel.basicConsume ( queue1Name , true , consumer );}
}

RabbitMQ路由模式:

特点：绑定了路由规则，交换机和队列再在定时需要绑定路由规则，生产者发送消息时也会绑定路由规则，消息会存放到路由规则匹配的队列中，只有符合路由规则的消费者才能取出消息

RabbitMQ通配符模式:

特点：在交换机和队列绑定时，通过设置通配符的方式来设置路由规则，当生产者发送信息时会绑定路由规则，只有符合规则的消费者才会消费信息，这种方式比Routing更灵活一些。