public class Producer {
private final static String QUEUE*NAME = "hello";
public static void main(String[] args) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
String ip = "192.168.117.128";
String username = "admin";
factory.setHost(ip);
factory.setUsername(username);
String password = "123456";
factory.setPassword(password);
try (Connection connection = factory.newConnection();
Channel channel = connection.createChannel()) {
/\*\*
* 生成一个队列
_ 1.队列名称
_ 2.队列里面的消息是否持久化 默认消息存储在内存中
_ 3.该队列是否只供一个消费者进行消费 是否进行共享 true 可以多个消费者消费
_ 4.是否自动删除 最后一个消费者端开连接以后 该队列是否自动删除 true 自动删除
_ 5.其他参数
\*/
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
String message;
while (true) {
Scanner scanner = new Scanner(System.in);
message = scanner.next();
/\*\*
_ 发送一个消息
_ 1.发送到那个交换机
_ 2.路由的 key 是哪个
_ 3.其他的参数信息
_ 4.发送消息的消息体
\_/
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println("message send success");
TimeUnit.MILLISECONDS.sleep(100);
}
}
}
}

public class Consumer {
private final static String QUEUE_NAME = "hello";

public static void main(String[] args) throws Exception {
ConnectionFactory connectionFactory = new ConnectionFactory();
String password = "123456";
String host = "192.168.117.128";
String username = "admin";
connectionFactory.setPassword(password);
connectionFactory.setHost(host);
connectionFactory.setUsername(username);
try (Connection connection = connectionFactory.newConnection();
Channel channel = connection.createChannel()) {
System.out.println("waiting from 2");
//推送的消息如何进行消费的接口回调
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody());
System.out.println(message);
System.out.println("received from c2");
};
//取消消费的一个回调接口 如在消费的时候队列被删除掉了
CancelCallback cancelCallback = (consumerTag) -> {
System.out.println("message consume was interrupted");
};
while (true) {
TimeUnit.SECONDS.sleep(1);
/\*\*
_ 消费者消费消息
_ 1.消费哪个队列
_ 2.消费成功之后是否要自动应答 true 代表自动应答 false 手动应答
_ 3.消费者成功的回调
_ 4.消费未成功的回调
_/
channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);
}
}
}
}

public class Consumer {
private final static String QUEUE_NAME = "hello";
static Channel channel;

static {
ConnectionFactory connectionFactory = new ConnectionFactory();
String password = "123456";
String host = "192.168.117.128";
String username = "admin";
connectionFactory.setPassword(password);
connectionFactory.setHost(host);
connectionFactory.setUsername(username);
try {
Connection connection = connectionFactory.newConnection();
channel = connection.createChannel();
} catch (IOException | TimeoutException e) {
e.printStackTrace();
}
}

public static void main(String[] args) throws Exception {
DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String receivedMessage = new String(delivery.getBody());
System.out.println("接收到消息:" + receivedMessage);
};

      CancelCallback cancelCallback = (consumerTag) -> {
         System.out.println(consumerTag + "消费者取消消费接口回调逻辑");
      };
      System.out.println("C1 消费者启动等待消费.................. ");
    	/**多个consumer公用一个 channel 否则并不会出现轮训分发情况*/
      channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);

}
}

DeliverCallback deliverCallback = (consumerTag, delivery) -> {
String message = new String(delivery.getBody(), "UTF-8");
System.out.println(" [x] Received '" + message + "'");
try {
//处理消息
doWork(message);
} finally {
System.out.println(" [x] Done");
/\*\*
_ 1. 消息标记 tag
_ 2. false 代表只应答接收到的那个传递消息，true 为应答所有传递过来的消息
\*/
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}
};
boolean autoAck = false; //手动应答
channel.basicConsume(TASK_QUEUE_NAME, autoAck, deliverCallback, consumerTag -> { });

int prefetchCount = 1 ;
channel.basicQos(prefetchCount);

boolean durable = true;
channel.queueDeclare(ACK_QUEUE_NAME, durable, false, false, null);

import com.rabbitmq.client.MessageProperties;

channel.basicPublish("", "task_queue",
            MessageProperties.PERSISTENT_TEXT_PLAIN,
            message.getBytes());