RabbitMQ的工作模式——WorkQueues

1.工作队列模式

生产者代码

public class Producer_WorkQueues1 {public static void main(String[] args) throws IOException, TimeoutException {//1.创建连接工厂ConnectionFactory factory = new ConnectionFactory();//2.设置参数factory.setHost("172.16.98.133"); ip 默认值 localhostfactory.setPort(5672);//端口 默认值5672factory.setVirtualHost("/itcast");//虚拟机 默认值factory.setUsername("heima");//用户名 默认guestfactory.setPassword("heima");//密码 默认值 guest//3.创建连接 ConnectionConnection connection = factory.newConnection();//4.创建ChannelChannel channel = connection.creatChannel();//5.创建队列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：参数。*///如果没有一个名字叫hello_world的队列，则会创建该队列，如果有则不会创建channel.queueDeclare("work_queues",true,false,false,null);/*basicPublish(String exchange,String routingKey,BasicProperties props, byte[] body)参数：1.exchange：交换机名称。简单模式下交换机默认使用的“”2.routingKey：路由名称3.props：配置信息4.body: 发送消息数据*/for(int i = 1; i<=10;i++){String body = i + "hello rabbitmq~~~";//发送消息channel.basicPublish("","work_queues",null,body.getBytes());}//7.释放资源channel.close();connection.close();}
}

消费之代码

public class Consumer_WrokQueues1 {public static void main(String[] args) throws IOException, TimeoutException {//1.创建连接工厂ConnectionFactory factory = new ConnectionFactory();//2.设置参数factory.setHost("172.16.98.133"); ip 默认值 localhostfactory.setPort(5672);//端口 默认值5672factory.setVirtualHost("/itcast");//虚拟机 默认值factory.setUsername("heima");//用户名 默认guestfactory.setPassword("heima");//密码 默认值 guest//3.创建连接 ConnectionConnection connection = factory.newConnection();//4.创建ChannelChannel channel = connection.creatChannel();//5.创建队列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：参数。*///如果没有一个名字叫hello_world的队列，则会创建该队列，如果有则不会创建channel.queueDeclare("work_queues",true,false,false,null);/*basicConsume(String queue,boolean autoAck,Consumer callback)参数：1.queue：队列名称2.autoAck：是否自动确认3.callback：回调对象*///接收消息Consumer consumer = new DefaultConsumer(channel){/*回调方法，当收到消息后会自动执行该方法1.consumerTag：标识2.envelope：获取一些信息，交换机，路由key...3.properties：配置信息4.body：数据*/@Overridepublic void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body){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("work_queues",true,consumer);//消费者不能关闭资源}
}

启动连个消费者从同一个生产者里面争抢消息。