MQ控制台简单操作
建立Exchange
新建Exchange成功
新建Queue
新建Queue成功
建立Exchange与Queue的关系
建立关系成功
路由键: 就是指发送到Exchange的消息, 通过路由键的匹配规则, 分发到指定的Queue
样例: test.*
说明: 代表该消息的发送时的路由Key必须是以test.开头, 才会被分发到queue-test这个队列
*: 代表任意
发送消息
发送成功
获取消息
获取消息成功, 获取不代表消费~
Java简单操作MQ
添加依赖
<!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client --> <dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>5.15.0</version> </dependency>
生产者
package com.dance.redis.mq.rabbit; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; import com.rabbitmq.client.MessageProperties; import java.io.IOException; import java.util.concurrent.TimeoutException; public class RabbitProducer { private static final String EXCHANGE_NAME = "exchange-test"; private static final String ROUTING_KEY = "text.*"; private static final String QUEUE_NAME = "queue-test"; private static final String IP_ADDRESS = "192.168.247.142"; private static final int PORT = 5672; //RabbitMQ服务默认端口为5672 public static void main(String[] args) throws IOException, TimeoutException, InterruptedException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost(IP_ADDRESS); factory.setPort(PORT); factory.setUsername("root"); factory.setPassword("123456"); Connection connection = factory.newConnection();//创建连接 Channel channel = connection.createChannel();//创建信道 //创建一个type="topic"、持久化的、非自动删除的交换器。 channel.exchangeDeclare(EXCHANGE_NAME, "topic", true, false, null); //创建一个持久化、非排他的、非自动删除的队列 channel.queueDeclare(QUEUE_NAME, true, false, false, null); //将交换机与队列通过路由键绑定 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY); //发送一条持久化消息:Hello World! String message = "Hello World!"; channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes()); //关闭资源 channel.close(); connection.close(); } }
消费者
package com.dance.redis.mq.rabbit; import com.rabbitmq.client.*; import java.io.IOException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; public class Receiver { private static final String QUEUE_NAME = "queue-test"; private static final String IP_ADDRESS = "192.168.247.142"; private static final int PORT = 5672; public static void main(String[] args) throws IOException, TimeoutException, InterruptedException { Address[] address = new Address[]{ new Address(IP_ADDRESS, PORT) }; ConnectionFactory factory = new ConnectionFactory(); factory.setUsername("root"); factory.setPassword("123456"); // 这里的连接方式与生产者的demo略有不同,注意区别。 Connection connection = factory.newConnection(address); //创建连接 final Channel channel = connection.createChannel();//创建信道 channel.basicQos(64);//设置客户端最多接收未被ack的消息个数 Consumer consumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println("recvive message:" + new String(body)); try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); } channel.basicAck(envelope.getDeliveryTag(), false); } }; channel.basicConsume(QUEUE_NAME, consumer); //等待回调函数执行完毕之后,关闭资源。 TimeUnit.SECONDS.sleep(5); channel.close(); connection.close(); } }
标签:Java,String,05,factory,RabbitMQ,static,import,final,channel From: https://www.cnblogs.com/flower-dance/p/16754773.html