Rabbit简单队列模式

 1 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

   <modelVersion>4.0.0</modelVersion>

   <groupId>com.kf</groupId>

   <artifactId>rabbitMQ.demo</artifactId>

   <version>0.0.1-SNAPSHOT</version>

   <dependencies>

         <dependency>

             <groupId>com.rabbitmq</groupId>

             <artifactId>amqp-client</artifactId>

             <version>3.6.5</version>

         </dependency>

     </dependencies>

 </project>

 package com.kf.utils;

 import java.io.IOException;

 import java.util.concurrent.TimeoutException;

 import com.rabbitmq.client.Connection;

 import com.rabbitmq.client.ConnectionFactory;

 /**

  * 获取rabbit

  * @author kf

  *

  */

 public class RabbitConnectionUtils {

     public static Connection getConnection() throws IOException, TimeoutException{

         ConnectionFactory factory = new ConnectionFactory();

         factory.setHost("127.0.0.1");

         factory.setUsername("admin");

         factory.setPassword("admin");

         //AMQP协议端口号

         factory.setPort(5672);

         factory.setVirtualHost("/kf");

         Connection newConnection = factory.newConnection();

         return newConnection;

     }

 }

 package com.kf.queueDemo.simpleQueue;

 import java.io.IOException;

 import java.util.concurrent.TimeoutException;

 import com.kf.utils.RabbitConnectionUtils;

 import com.rabbitmq.client.Channel;

 import com.rabbitmq.client.Connection;

 /**

  * 简单队列

  * @author kf

  *

  */

 public class SimpleQueueProducer {

     //队列名称

     private static String QUEUENAME = "SIMPLEQUEUE";

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

         Connection connection = RabbitConnectionUtils.getConnection();

         //创建通道

         Channel channel = connection.createChannel();

         //通道里放入队列

         /**

          * 第一个参数是  队列名称

          * 第二个参数指 要不要持久化

          */

         channel.queueDeclare(QUEUENAME, false, false, false, null);

 /*        //消息体

         String mes = "demo_message汉字";

         //发送消息

         *//**

          * 参数为  exchange, routingKey, props, body

          * exchange   交换机

          * routingKey 路由键

          *

          * body 消息体

          *//*

         channel.basicPublish("", QUEUENAME, null, mes.getBytes());*/

         /**

          * 集群环境下，多个消费者情况下。消费者默认采用均摊

          */

         for(int i=1; i<11; i++){

             String mes = "demo_message汉字"+i;

             System.out.println("发送消息"+mes);

             channel.basicPublish("", QUEUENAME, null, mes.getBytes());

         }

 //        System.out.println("发送消息"+mes);

         channel.close();

         connection.close();

     }

 }

 package com.kf.queueDemo.simpleQueue;

 import java.io.IOException;

 import java.util.concurrent.TimeoutException;

 import com.kf.utils.RabbitConnectionUtils;

 import com.rabbitmq.client.Channel;

 import com.rabbitmq.client.Connection;

 import com.rabbitmq.client.DefaultConsumer;

 import com.rabbitmq.client.Envelope;

 import com.rabbitmq.client.AMQP.BasicProperties;

 /**

  * 简单队列消费者

  * @author kf

  *

  */

 public class SimpleConsumer {

     //队列名称

         private static String QUEUENAME = "SIMPLEQUEUE";

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

             System.out.println("01开始接收消息");

             Connection connection = RabbitConnectionUtils.getConnection();

             //创建通道

             final Channel channel = connection.createChannel();

             //通道里放入队列

             /**

              * 第一个参数是  队列名称

              * 第二个参数指 要不要持久化

              */

             channel.queueDeclare(QUEUENAME, false, false, false, null);

             DefaultConsumer consumer = new DefaultConsumer(channel){

                 //监听队列

                     @Override

                     public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,

                             byte[] body) throws IOException {

                         System.out.println("------------进入监听---------");

                         String s = new String(body, "utf-8");

                         System.out.println("获取到的消息是:"+s);

                         //手动应答。

                         /**

                          * 当  channel.basicConsume(QUEUENAME, true, consumer);第二个参数为false时  是手动应答模式

                          */

                     //    channel.basicAck(envelope.getDeliveryTag(), false);

                     }

             };

             //设置应答模式

             /**

              * 参数：  对列名，是否自动签收，监听的类

              */

             System.out.println("获取消息的方法之前");

             channel.basicConsume(QUEUENAME, true, consumer);

             System.out.println("获取消息的方法之后");

         }

 }