RabbitMQ 核心部分之简单模式和工作模式

publicclassProducer {
privatefinalstaticStringQUEUE_NAME="hello";
publicstaticvoidmain(String[] args) throwsException {
//创建一个连接工厂ConnectionFactoryfactory=newConnectionFactory();
factory.setHost("192.168.10.130");
factory.setUsername("guest");
factory.setPassword("guest");
//channel 实现了自动 close 接口 自动关闭 不需要显示关闭try(Connectionconnection=factory.newConnection();Channelchannel=connection.createChannel()) {
/*** 生成一个队列* 1.队列名称* 2.队列里面的消息是否持久化 默认消息存储在内存中* 3.该队列是否只供一个消费者进行消费 是否进行共享 true 可以多个消费者消费* 4.是否自动删除 最后一个消费者端开连接以后 该队列是否自动删除 true 自动删除* 5.其他参数*/channel.queueDeclare(QUEUE_NAME,false,false,false,null);
Stringmessage="hello world";
/*** 发送一个消息* 1.发送到那个交换机* 2.路由的 key 是哪个* 3.其他的参数信息* 4.发送消息的消息体*/channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
System.out.println("消息发送完毕");
        }
    }
}

publicclassConsumer {
privatefinalstaticStringQUEUE_NAME="hello";
publicstaticvoidmain(String[] args) throwsException {
ConnectionFactoryfactory=newConnectionFactory();
factory.setHost("192.168.10.130");
factory.setUsername("guest");
factory.setPassword("guest");
Connectionconnection=factory.newConnection();
Channelchannel=connection.createChannel();
System.out.println("等待接收消息....");
//推送的消息如何进行消费的接口回调DeliverCallbackdeliverCallback=(consumerTag,delivery)->{
Stringmessage=newString(delivery.getBody());
System.out.println(message);
        };
//取消消费的一个回调接口 如在消费的时候队列被删除掉了CancelCallbackcancelCallback=(consumerTag)->{
System.out.println("消息消费被中断");
        };
/*** 消费者消费消息* 1.消费哪个队列* 2.消费成功之后是否要自动应答 true 代表自动应答 false 手动应答* 3.消费者未成功消费的回调*/channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
    }
 }

publicclassRabbitMqUtils {
//得到一个连接的 channelpublicstaticChannelgetChannel() throwsException{
//创建一个连接工厂ConnectionFactoryfactory=newConnectionFactory();
factory.setHost("192.168.10.130");
factory.setUsername("guest");
factory.setPassword("guest");
Connectionconnection=factory.newConnection();
Channelchannel=connection.createChannel();
returnchannel;
    }
}

publicclassWorker01 {
privatestaticfinalStringQUEUE_NAME="hello";
publicstaticvoidmain(String[] args) throwsException {
Channelchannel=RabbitMqUtils.getChannel();
DeliverCallbackdeliverCallback=(consumerTag,delivery)->{
StringreceivedMessage=newString(delivery.getBody());
System.out.println("接收到消息:"+receivedMessage);
        };
CancelCallbackcancelCallback=(consumerTag)->{
System.out.println(consumerTag+"消费者取消消费接口回调逻辑");
        };
System.out.println("C2 消费者启动等待消费......");
channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
    }
}

publicclassTask01 {
privatestaticfinalStringQUEUE_NAME="hello";
publicstaticvoidmain(String[] args) throwsException {
try(Channelchannel=RabbitMqUtils.getChannel();) {
channel.queueDeclare(QUEUE_NAME,false,false,false,null);
//从控制台当中接受信息Scannerscanner=newScanner(System.in);
while (scanner.hasNext()){
Stringmessage=scanner.next();
channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
System.out.println("发送消息完成:"+message);
            }
        }
    }
}