rabbitmq 基本总结

rabbitmq 的基本概念 vhost、broker、producer、 consumer、 exchange、 queue、 routing key

rabbitmq 常用的队列类型，工作队列（简单队列）,pub/sub, routing key， topic 模式

<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.16.0</version>
</dependency>

public class RabbitmqClientDemo {private static ConnectionFactory factory = new ConnectionFactory();private static String EXCHANGE_NAME = "exchange.fanout";private static String FANOUT_QUEUE = "queue.fanout";private static String DIRECT_EXCHANGE = "exchange_direct";private static String QUEUE_DIRCT = "queue.direct.02";private static String QUEUE_TOPIC_ONE = "queue.topic.01";private static String QUEUE_TOPIC_TWO = "queue.topic.02";private static String QUEUE_TOPIC_THREE = "queue.topic.03";private static String ROUNTING_KEY_ONE = "routing.key.01";private static String ROUNTING_KEY_TWO = "routing.key.02";private static String ROUNTING_KEY_THREE = "routing.key.03";private static String DEAD_MESSAGE_EXCHANGE = "EXCHANGE_DEAD";private static String DEAD_QUEUE = "queue.dead";static {factory.setHost("192.168.233.128");factory.setPort(5672);factory.setUsername("guest");factory.setPassword("guest");}public static Connection getConnection() throws IOException, TimeoutException {return factory.newConnection();}public static Channel createChannel() throws IOException, TimeoutException {Connection connection = getConnection();return connection.createChannel();}public static void main(String[] args) {//new WorkQueueProducer().start();//new WorkerConsumer().start();/*  new PublishConsumer().start();new PublishProducer().start();*///new TopicProducer().start();//new TopicConsumer().start();new DeadMessageProducer().start();new DeadMessageConsumer().start();}static class WorkQueueProducer extends Thread {@Overridepublic void run() {try {Connection connection = getConnection();Channel channel = connection.createChannel();channel = connection.createChannel();channel.queueDeclare("hello", false, false, false, null);channel.basicPublish("", "hello", null, "hello".getBytes());} catch (Exception e) {e.printStackTrace();} finally {try {//hannel.close();} catch (Exception e) {e.printStackTrace();}}}}static class WorkerConsumer extends Thread {@Overridepublic void run() {try {Connection connection = getConnection();Channel channel = connection.createChannel();channel.queueDeclare("hello", false, false, false, null);channel.basicQos(1);channel.basicConsume("hello", true, new DeliverCallback() {@Overridepublic void handle(String s, Delivery delivery) throws IOException {System.out.println(new String(delivery.getBody()));}}, consumerTag -> {});} catch (Exception e) {e.printStackTrace();} finally {try {//channel.close();} catch (Exception e) {e.printStackTrace();}}}}static class PublishProducer extends Thread {@Overridepublic void run() {try {Channel channel = createChannel();channel.exchangeDeclare(EXCHANGE_NAME, "fanout");channel.queueDeclare(FANOUT_QUEUE, true, false, false, null);channel.queueBind(FANOUT_QUEUE, EXCHANGE_NAME, "");for (int i = 1; i <= 40; i++) {String message = String.format("current orderId is %d, money is %d", UUID.randomUUID(), new Random().nextDouble());channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes(StandardCharsets.UTF_8));}} catch (Exception e) {e.printStackTrace();} finally {try {//channel.close();} catch (Exception e) {e.printStackTrace();}}}}static class PublishConsumer extends Thread {@Overridepublic void run() {try {Channel channel = createChannel();channel.exchangeDeclare(EXCHANGE_NAME, "fanout");channel.queueDeclare(FANOUT_QUEUE, true, false, false, null);channel.queueBind(FANOUT_QUEUE, EXCHANGE_NAME, "");DeliverCallback deliverCallback = (consumerTag, delivery) -> {System.out.println(new String(delivery.getBody(), StandardCharsets.UTF_8));};while (true) {channel.basicConsume(FANOUT_QUEUE, true, deliverCallback, consumerTag -> {});}} catch (Exception e) {e.printStackTrace();} finally {try {//channel.close();} catch (Exception e) {e.printStackTrace();}}}}static class TopicProducer extends Thread {@Overridepublic void run() {try {Channel channel = createChannel();channel.exchangeDeclare(DIRECT_EXCHANGE, "topic");channel.queueDeclare(QUEUE_TOPIC_ONE, true, false, false, null);channel.queueDeclare(QUEUE_TOPIC_TWO, true, false, false, null);channel.queueDeclare(QUEUE_TOPIC_THREE, true, false, false, null);channel.queueBind(QUEUE_TOPIC_ONE, DIRECT_EXCHANGE, ROUNTING_KEY_ONE);channel.queueBind(QUEUE_TOPIC_TWO, DIRECT_EXCHANGE, ROUNTING_KEY_TWO);channel.queueBind(QUEUE_TOPIC_TWO, DIRECT_EXCHANGE, ROUNTING_KEY_TWO);channel.queueBind(QUEUE_TOPIC_THREE, DIRECT_EXCHANGE, ROUNTING_KEY_THREE);channel.queueBind(QUEUE_TOPIC_THREE, DIRECT_EXCHANGE, ROUNTING_KEY_ONE);channel.queueBind(QUEUE_TOPIC_THREE, DIRECT_EXCHANGE, ROUNTING_KEY_TWO);for (int i = 1; i <= 10; i++) {String message = String.format("current orderId is %s, money is %s", UUID.randomUUID().toString(), new Random().nextDouble());if (i % 3 == 0) {System.out.println("send to topic1");channel.basicPublish(DIRECT_EXCHANGE, ROUNTING_KEY_ONE, null, message.getBytes(StandardCharsets.UTF_8));} else if (i % 3 == 1) {System.out.println("send to topic2");channel.basicPublish(DIRECT_EXCHANGE, ROUNTING_KEY_TWO, null, message.getBytes(StandardCharsets.UTF_8));} else {System.out.println("send to topic3");channel.basicPublish(DIRECT_EXCHANGE, ROUNTING_KEY_THREE, null, message.getBytes(StandardCharsets.UTF_8));}}} catch (Exception e) {e.printStackTrace();}}}static class TopicConsumer extends Thread {@Overridepublic void run() {try {Channel channel = createChannel();channel.exchangeDeclare(DIRECT_EXCHANGE, "topic");channel.queueDeclare(QUEUE_TOPIC_THREE, true, false, false, null);channel.queueBind(QUEUE_TOPIC_THREE,EXCHANGE_NAME,"routing.key.*")DeliverCallback deliverCallback = (consumerTag, delivery) -> {System.out.println(delivery.getEnvelope().getRoutingKey());System.out.println(new String(delivery.getBody(), StandardCharsets.UTF_8));};while (true) {channel.basicConsume(QUEUE_TOPIC_THREE, true, deliverCallback, consumerTag -> {});}} catch (Exception e) {e.printStackTrace();} finally {try {//channel.close();} catch (Exception e) {e.printStackTrace();}}}}static class DeadMessageProducer extends Thread {@Overridepublic void run() {try {Channel channel = createChannel();channel.exchangeDeclare(DIRECT_EXCHANGE, "direct");channel.queueDeclare(DEAD_QUEUE, true, false, false, null);channel.queueBind(DEAD_QUEUE, DEAD_MESSAGE_EXCHANGE, "routing.direct02");for (int i = 1; i <= 40; i++) {String message = String.format("current orderId is %s, money is %s", UUID.randomUUID().toString(), new Random().nextDouble());AMQP.BasicProperties prop = new AMQP.BasicProperties().builder().expiration("30000").build();channel.basicPublish(DIRECT_EXCHANGE, "routing.direct02", prop, message.getBytes(StandardCharsets.UTF_8));System.out.println("send to topic1");}} catch (Exception e) {e.printStackTrace();}}}static class DeadMessageConsumer extends Thread {@Overridepublic void run() {try {Channel channel = createChannel();/*channel.exchangeDeclare(DEAD_MESSAGE_EXCHANGE, "direct");Map<String, Object> deadLetterParams = new HashMap<>(2);deadLetterParams.put("x-dead-letter-exchange", DEAD_MESSAGE_EXCHANGE);deadLetterParams.put("x-dead-letter-routing-key", "routing.dead02");deadLetterParams.put("x-max-length", 2);*//*channel.queueDeclare(QUEUE_DIRCT, true, false, false, deadLetterParams);channel.queueBind(DEAD_QUEUE, DEAD_MESSAGE_EXCHANGE, "routing.dead02");*/channel.exchangeDeclare(DIRECT_EXCHANGE, "direct");channel.queueBind(QUEUE_DIRCT, DIRECT_EXCHANGE, "routing.direct02");DeliverCallback callback  = (consumerTag, delivery) -> {System.out.println(delivery.getEnvelope().getRoutingKey());System.out.println(new String(delivery.getBody(), StandardCharsets.UTF_8));};/* DeliverCallback callback = (consumerTag, delivery) -> {String receivedMessage = new String(delivery.getBody());System.out.println("C1接收到消息：" + receivedMessage + "并且拒绝签收了");// 禁止重新入队channel.basicReject(delivery.getEnvelope().getDeliveryTag(), false);};*/while (true) {//channel.basicConsume(QUEUE_DIRCT, true, deliverCallback, consumerTag -> {});channel.basicConsume(QUEUE_DIRCT, true, callback, (consumerTag) -> {System.out.println(consumerTag + "消费者取消消费消息");});}} catch (Exception e) {e.printStackTrace();} finally {try {//channel.close();} catch (Exception e) {e.printStackTrace();}}}}

整合springboot

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId><version>3.1.5</version>
</dependency>

rabbitmq 的核心配置（相比于其他的mq，rabbit 有图形用户界面，可以傻瓜操作）

https://blog.csdn.net/leesinbad/article/details/128670794