一、第一章

1、pom依赖

二、第二章

1、消息属性对象(Delivery delivery)

2、信道对象 (发送消息根据路由发送，接收消息根据队列接收)

3、工作队列模式

4、消息应答 (消费者)

5、消息自动重新入队 (消费者)

6、RabbitMQ持久化 (生产者)

7、不公平分发(能者多劳)

8、发布确认消息(生产者)

9、交换机

10、Exchanges交换机的类型

11、临时队列

12、交换机和队列的关系

13、交换机模式

（1）Fanout模式(发布/订阅)

（2）Direct 直接交换机模式(路由模式)

（3）Topic 主题模式

14、死信队列(队列消息自动发送到配置的路由中)

15、延迟队列(ttl)

一、第一章

1、pom依赖

   <dependencies><!--rabbitmq 依赖客户端--><dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.8.0</version></dependency><!--操作文件流的一个依赖--><dependency><groupId>commons-io</groupId><artifactId>commons-io</artifactId><version>2.6</version></dependency></dependencies><!--指定 jdk 编译版本--><build><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><configuration><source>8</source><target>8</target></configuration></plugin></plugins></build>

二、第二章

1、消息属性对象(Delivery delivery)

Delivery 类

api

   delivery.getBody() 消息的内容(byte类型)   装箱: new String(delivery.getBody())delivery.getProperties() 当前消息的详细属性(很多get方法)delivery.getEnvelope() 消息的属性delivery.getEnvelope().getDeliveryTag() 当前消息在当前消费者中的位置 (1标识当前消费者处理的第一条消息)delivery.getEnvelope().getRoutingKey()  消息的路由Keydelivery.getEnvelope().getExchange();  消息的交换机delivery.getEnvelope().isDedeliver(); 是否是重新发送消息

2、信道对象 (发送消息根据路由发送，接收消息根据队列接收)

//创建连接
Connection connection=factory.newConnection();

//发消息是通过信道发送消息，所以要创建信道
//获取信道
Channel channel= connection.createChannel();

   消费者channel.basicAck(delivery,true)       (用于肯定确认应答) true表示批量应答信道中的消息都会被应答 (未被处理的也应答),false表示不批量应答，只应答当前消息channel.basicNack       (用于否定确认应答)channel.basicReject     (用于否定确认应答)channel.basicQos(1)   (消息暂存区大小，0无限大，n表示当前有消息未处理完时候可以容纳的其他消息条数)//队列    //自动应答  //成功回调      // 消息被取消的回调channel.basicConsume(QUEUE_NAME,true, deliverCallback1,cancelCallback); //(消息是异步的) 不会阻塞当前线程的运行

     生产者channel.confirmSelect();  //开启发布确认，返回值为bool//设置队列名   持久化  单个消费者  自动删除   其他参数channel.queueDeclare("cons" ,false    ,false,     false,      null);  //创建队列 创建在生产者或消费者都可String queueName=channel.queueDedare.getQueue(); //，创建一个临时队列，返回值队列的名称//交换机  路由id    其他参数     消息内容channel.basicPublish("",QUEUE_NAME,null,mssage.getBytes()); //发送消息boolean flag=channel.waitForConfirms(); // 确认前面全部发布的消息 （可以每发布调用一次，也可以批量发布后再确认）//确认消息回调          未确认消息消息回调channel.addConfirmListener(confirmCallback1,confirmCallback2); // 异步应答监听  (需要创建再发布消息之前)

  其他channel.getNextPublishSeqNo() //下一次发送的消息的标号 (标号从1开始)//声名一个交换机           //交换机名字    交换机类型channel.exchangeDeclare(“logs”,       BuiltinExchangeType.FANOUT );        //定义在消费者或者生产者都行 创建在生产者或消费者都可//绑定交换机和队列         队列名   交换机名       routingKey 绑定序号(路由序号)channel.queueBind(queueName,   "logs",   "");       //定义在消费者或者生产者都行

示例代码

连接工厂

//连接工厂创建信道的工具类
public class RabbitMqUtis {public static Channel getChannel() throws Exception{//创建连接工厂ConnectionFactory factory=new ConnectionFactory();//工厂ip 连接 RabbitMQ的队列factory.setHost("192.168.1.5");//用户名factory.setUsername("usermq");//密码factory.setPassword("111111");//创建连接Connection connection=factory.newConnection();//  MessageProperties.PERSISTENT_TEXT_PLAIN//发消息是通过信道发送消息，所以要创建信道//获取信道return  connection.createChannel();}
}

生产者


//生产者
//发送大量的消息
//发消息
public class Task_one {//队列名称public static final  String QUEUE_NAME="hello";//发送大量消息public static void main(String[] args) throws Exception {Channel channel=RabbitMqUtis.getChannel();//声明一个队列//参数//1、队列名称//2、队列里的消息是否持久化(磁盘) 默认情况消息存储在内存中//3、该队列是否只提供一个消费者进行消费 是否进行消息共享，true可以多个消费者消费//4、是否自动删除 最后一个消费者端开连接以后 该队一句是否删除 true自动删除 false不自动删除//5、其他参数channel.queueDeclare(QUEUE_NAME,false,false,false,null);//从控制台接收消息Scanner scanner=new Scanner(System.in);while (scanner.hasNext()){String mssage=scanner.next();//1、发送到哪个交换机//2、路由的key值是哪个，本次是队列的名称//3、消息持久化(MessageProperties.PERSISTENT_TEXT_PLAIN)//4、发送消息的消息体channel.basicPublish("",QUEUE_NAME,null,mssage.getBytes());System.out.println("发送消息完成："+mssage);}}
}

消费者


//消费者工作线程
// 这是一个线程
public class Worker_one {//队列的名称public static final  String QUEUE_NAME="hello";//接收消息public static void main(String[] args) throws Exception {//获得信道Channel channel= RabbitMqUtis.getChannel();//消费消息//消费者接收消息//1、消费哪个队列//2、消费成功之后是否要自动应答 (true代表自动应答,false代表手动应答)//3、消费者消费成功的回调方法 (拉姆达表达式)//4、消费者消费消费失败的回调  (拉姆达表达式)//成功回调DeliverCallback deliverCallback1=new DeliverCallback() {@Overridepublic void handle(String s, Delivery delivery) throws IOException {//  s:当前会话 (同一个消费者会话是相对的)//  delivery 消息属性// new String( delivery.getBody() 消息的valueSystem.out.println("s:"+s);System.out.println("delivery.getBody():"+ new String( delivery.getBody()));System.out.println("delivery.getEnvelope().getDeliveryTag():"+delivery.getEnvelope().getDeliveryTag());System.out.println("delivery.getExchange().getDeliveryTag():"+delivery.getEnvelope().getExchange());System.out.println("delivery.getExchange().)getRoutingKey():"+delivery.getEnvelope().getRoutingKey());}};//消息被取消CancelCallback cancelCallback=new CancelCallback() {@Overridepublic void handle(String s) throws IOException {System.out.println("当前会话："+s);}};System.out.println("C2等待接收消息....");//队列    //自动应答  //成功回调      // 消息被取消的回调channel.basicConsume(QUEUE_NAME,true, deliverCallback1,cancelCallback);}
}

模式

  p为生产者  c为消费者  在中间的队列 rabbitmqp生产者(java程序)创建连接工厂--通过工厂创建连接--通过连接创建信道--设置信道队列--发送消息代码..//声明一个队列//参数//1、队列名称//2、队列里的消息是否持久化(磁盘) 默认情况消息存储在内存中//3、该队列是否只提供一个消费者进行消费 是否进行消息共享，true可以多个消费者消费//4、是否自动删除 最后一个消费者端开连接以后 该队列是否删除 true自动删除 false不自动删除//5、其他参数//队列   持久化  单个消费者  自动删除     其他配置(map)channel.queueDeclare(QUEUE_NAME ,false  ,false,   false,      null);//1、发送到哪个交换机//2、路由的key值是哪个，本次是队列的名称//3、其他参数信息//4、发送消息的消息体//交换机 路由的key  其他配置(AMQP)    消息内容channel.basicPublish("",   QUEUE_NAME,  null,   mssage.getBytes());

       c为消费者(java程序)创建连接工厂--通过工厂创建连接--通过连接创建信道--通过信道接收消息执行消费者语句是异步的，它有自己的独立线程代码..//消费消息//消费者接收消息//1、消费哪个队列//2、消费成功之后是否要自动应答 (true代表自动应答,false代表手动应答)//3、消费者消费成功的回调方法 (拉姆达表达式)//4、消费者消费消费失败的回调  (拉姆达表达式)//成功回调DeliverCallback deliverCallback1=new DeliverCallback() {@Overridepublic void handle(String s, Delivery delivery) throws IOException {//  s:当前会话 (同一个消费者会话是相同的)//  delivery 消息属性// new String( delivery.getBody() 消息的value//delivery.getEnvelope() 消息的属性//delivery.getEnvelope().getDeliveryTag() 消息的标记// delivery.getBody() 消息的内容(比特数据)System.out.println("s:"+s);System.out.println("delivery.getBody():"+ new String( delivery.getBody(),"UTF-8"));}};//消息被取消CancelCallback cancelCallback=new CancelCallback() {@Overridepublic void handle(String s) throws IOException {System.out.println("当前会话："+s);}};//队列    //自动应答  //成功回调      // 消息被取消的回调channel.basicConsume(QUEUE_NAME,true, deliverCallback1,cancelCallback); //(消息是异步的) 不会阻塞当前线程的运行

3、工作队列模式

多线程处理消息:一条消息只能被处理一次
生产者发送消息----> 队列---> 接收消息大量工作线程
轮训接收消息
消费者工作线程轮流接收消息(工作线程是竞争关系)

4、消息应答 (消费者)

为了保证消息在发送过程中不丢失，rabbitmq引入消息应答机制：
如果未被应答会的时候消费者程序down掉(用于否定确认应答也是应答), 消息会重新返回队列发放给其他消费者

  (消费者 basic.reject或 basic.nack)并且requeue=false收到消息并且处理该消息之后，告诉rabbitmq它已经处理了，rabbitmq可以把该消息删除了自动应答：消费者收到消息就是成功手动应答：调用应答方法告诉RabbitMQ 该消息已经被处理了,如果未被应答会,消息会重新返回队列发放给其他消费者channel.basicAck        (用于肯定确认应答) channel.basicNack       (用于否定确认应答)channel.basicReject     (用于否定确认应答)与channel.basicNack相比少一个参数不处理该消息直接拒绝，可以将其丢弃了手动应答的好处是可以批量应答且减少网络拥堵，也可以重新放回队列不丢失数据批量应答channel.basicAck(message.getEnvelope().getDeliveryTag(),  true)message.getEnvelope().getDeliveryTag() ;当前消费者处理的消息的标识(从1开始)true表示批量应答信道中的消息都会被应答 (未被处理的也应答)false表示不批量应答，只应答当前消息channel.basicNack(message.getEnvelope().getDeliveryTag(),  true, true)message.getEnvelope().getDeliveryTag() ;当前消费者处理的消息的标识(从1开始)第一个true ：表示当前消息可以重新入队第二个true: 表示批量处理未确认的消息channel.basicReject(message.getEnvelope().getDeliveryTag(),  true); message.getEnvelope().getDeliveryTag():当前消费者处理的消息的标识(从1开始)true 表示否定确认的消息允许返回队列false   表示否定确认的消息不允许返回队列1、   //手动应答(正确应答) //1、消息的标记 tag (相当于当前消息在信道的位置)//2、是否批量应答 (批量应答会)//message.getEnvelope() 消息的属性//message.getEnvelope().getDeliveryTag() 消息的标识// message.getBody() 消息的内容channel.basicAck(message.getEnvelope().getDeliveryTag(),false);