作者：小池子的思密达 | 来源：互联网 | 2023-05-19 12:12
概念：RabbitMQ是一款开源的消息中间件系统，由erlang开发，是AMQP的实现。架构图大概如上。broker是消息队列的服务器，比如在linux上，我们安装的rabbitmq就是一
概念：RabbitMQ是一款开源的消息中间件系统，由erlang开发，是AMQP的实现。
架构图大概如上。
broker是消息队列的服务器，比如在linux上，我们安装的rabbitmq就是一个broker，可以通过url+username+password连接。
每个消息服务器可以创建多个vhost，默认的vhost是“/”，linux中通过rabbitmqctl add_vhost 创建vhost，再给指定用户授权即可。
生产者首先通过创建channel与broker连接，类似于创建一个会话，这样可以与消息主机通信发送消息。
消息生产者将消息发送到定义的exchange上，exchange通过不同的转发路由规则将消息转发到相应的队列，消费者选择一个队列监听，如果有多个消费者监听同一个队列，默认是轮询方式，保证每个连接有相同的收到消息的概率。
一个简单的rabbitmq程序:
public class Producer {
private static final String TEST_VHOST = "testvhost";
private static final String TEST_QUEUE_NAME = "task_queue";
private static Connection connection;
private static Channel channel;
public static void main(String[] args) throws IOException, TimeoutException, RabbitmqConnectionException {
try {
//create connectionFactory with host, username, password, vhost.
ConnectionFactory cOnnectionFactory= new ConnectionFactory();
connectionFactory.setUsername("test");
connectionFactory.setPassword("test");
connectionFactory.setHost("localhost");
connectionFactory.setVirtualHost(TEST_VHOST);
//get connection from connectionFactory
cOnnection= connectionFactory.newConnection();
//create an session to communicate with mq host
channel = connection.createChannel();
//declare a queue(if not exists, create it)
channel.queueDeclare(TEST_QUEUE_NAME, true, false, false, null);
String message = "Hello world";
System.out.println("sending message : " + message);
//publish message to the declaring queue
channel.basicPublish("", TEST_QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
}catch (Exception e) {
throw new RabbitmqConnectionException("Error connection");
} finally {
channel.close();
connection.close();
}
}
}
public class Consumer {
private static final String TEST_VHOST = "testvhost";
private static final String TEST_QUEUE_NAME = "task_queue";
public static void main(String[] argv) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
factory.setVirtualHost(TEST_VHOST);
factory.setUsername("test");
factory.setPassword("test");
final Connection cOnnection= factory.newConnection();
final Channel channel = connection.createChannel();
//declaring a queue to listen
channel.queueDeclare(TEST_QUEUE_NAME, true, false, false, null);
System.out.println("Waiting for messages...");
//a piece message per time
channel.basicQos(1);
final com.rabbitmq.client.Consumer cOnsumer= new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
String message = new String(body, "UTF-8");
System.out.println("Received : '" + message + "'");
channel.basicAck(envelope.getDeliveryTag(), false);
}
};
channel.basicConsume(TEST_QUEUE_NAME, false, consumer);
}
}
在spring中:
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xmlns:mvc="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
rabbitmq 连接服务配置
package="com.battery.rabbitMq"/>
class="org.springframework.amqp.support.converter.Jackson2JsonMessageConverter" />
class="com.battery.rabbitMq.QueueListener"/>
@Service
public class MQProducerImpl implements MQProducer {
@Autowired
private RabbitTemplate rabbitTemplate;
private final static Logger LOGGER = LoggerFactory.getLogger(MQProducerImpl.class);
/**
* convertAndSend：将Java对象转换为消息发送到匹配Key的交换机中Exchange，由于配置了JSON转换，这里是将Java对象转换成JSON字符串的形式。
* 原文：Convert a Java object to an Amqp Message and send it to a default exchange with a specific routing key.
**/
@Override
public void sendDataToQueue(Object object) {
try {
rabbitTemplate.convertAndSend(object);
} catch (Exception e) {
LOGGER.error(e.getMessage());
}
}
}
@Component
public class QueueListener implements ChannelAwareMessageListener {
private static Logger logger = LoggerFactory.getLogger(QueueListener.class);
@Override
public void onMessage(Message message, Channel channel) throws Exception {
try {
String ackMessage = new String(message.getBody(), "utf-8");
System.out.print(ackMessage);
logger.debug("接收到:" + new String(message.getBody(), "utf-8"));
} catch (Exception e) {
System.out.print(e.getMessage());
}
}
}
![[译]技术公司十年经验的职场生涯回顾](https://img8.php1.cn/3cdc5/24912/711/b6574f3292f9dc00.png)
京公网安备 11010802041100号