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);//消费者不能关闭资源}
}
启动连个消费者从同一个生产者里面争抢消息。