rabbit-1简介

女爷i 2022-05-17 06:47 137阅读 0赞

**消息（message）是指在应用件传递的数据。**

**消息队列中间件（message queue middleware MQ）**是指利用高效可靠的消息传递机制进行与平台无关的数据交流。并基于数据通信来进行分布式系统的集成。也称为消息队列或者消息中间件；一般有两种传递模式:点对点（P2P）和发布/订阅（pub/sub）模式。P2P是基于队列的；PUB/SUB定义了如何向一个内容节点发布和订阅消息，这个内容节点称为主题（topic）。

**消息中间件作用：**

解耦、存储、扩展性、削峰、可恢复性、顺序保证、缓存、异步通信

RabbitMQ采用Erlang语言实现AMQP(advanced message queuing protocol,高级消息队列)的消息中间件。

**Rabbit的具体特点：**

*  可靠性：abbitMQ使用一些机制来保证可靠性，如持久化、传输确认及发布确认等。
 *  灵活的路由
 *  扩展性
 *  高可用性
 *  多种协议
 *  多语言客户端
 *  管理界面
 *  插件机制

**创建Hello:**

POM.xml

<project xmlns="http://maven.apache.org/POM/4.0.0"
    	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    	<modelVersion>4.0.0</modelVersion>
    	<groupId>frabbitmq</groupId>
    	<artifactId>frabbitmq</artifactId>
    	<version>0.0.1-SNAPSHOT</version>
    	
    	<dependencies>
    		<dependency>
    			<groupId>com.rabbitmq</groupId>
    			<artifactId>amqp-client</artifactId>
    			<version>4.1.0</version>
    		</dependency>
    	</dependencies>
    
    </project>

生产者：

package rabbit.inaction.recv;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.MessageProperties;
    
    public class RabbitProducer {
    	private static final String EXCHANGE_NAME = "exchange_demo";
    	private static final String ROUTING_KEY = "routingkey_demo";
    	private static final String QUEUE_NAME = "queue_demo";
    	private static final String IP_ADDRESS = "192.168.17.129";
    	private static final int PORT = 5672;
    
    	public static void main(String[] args) throws IOException, TimeoutException {
    		ConnectionFactory factory = new ConnectionFactory();
    		factory.setHost(IP_ADDRESS);
    		factory.setPort(PORT);
    		factory.setUsername("root");
    		factory.setPassword("root123");
    		Connection connection = factory.newConnection();// 创建连接
    		Channel channel = connection.createChannel();// 创建信道
    		// 创建一个type = "direct",持久的、非自动删除的交换器
    		channel.exchangeDeclare(EXCHANGE_NAME, "direct", true, false, null);
    		// 创建一个持久化的，非排他的、非自动删除的队列
    		channel.queueDeclare(QUEUE_NAME, true, false, false, null);
    		// 将交换器与队列通过路由器绑定
    		channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);
    		// 发送一条持久化的消息
    		String message = "hello rabbit";
    		channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, MessageProperties.PERSISTENT_TEXT_PLAIN,
    				message.getBytes("UTF-8"));
    
    		channel.close();
    		connection.close();
    	}
    }

消费者：

package rabbit.inaction.recv;
    
    import java.io.IOException;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.TimeoutException;
    
    import com.rabbitmq.client.AMQP;
    import com.rabbitmq.client.Address;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.Consumer;
    import com.rabbitmq.client.DefaultConsumer;
    import com.rabbitmq.client.Envelope;
    
    public class RabbitConsumer {
    	private static final String EXCHANGE_NAME = "exchange_demo";
    	private static final String ROUTING_KEY = "routingkey_demo";
    	private static final String QUEUE_NAME = "queue_demo";
    	private static final String IP_ADDRESS = "192.168.17.129";
    	private static final int PORT = 5672;
    
    	public static void main(String[] args) {
    		Address[] addresses = new Address[] { new Address(IP_ADDRESS, PORT) };
    		ConnectionFactory factory = new ConnectionFactory();
    		factory.setUsername("root");
    		factory.setPassword("root123");
    		// 和创建生产者有所不同
    		try {
    			Connection connection = factory.newConnection(addresses);// 新建连接
    			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("recv  message:  " + new String(body));
    					try {
    						TimeUnit.SECONDS.sleep(1);
    					} catch (InterruptedException e) {
    						e.printStackTrace();
    						channel.basicAck(envelope.getDeliveryTag(), false);
    					}
    				}
    			};
    
    			channel.basicConsume(QUEUE_NAME, consumer);
    
    //			channel.close();
    //			connection.close();
    		} catch (IOException | TimeoutException e) {
    			e.printStackTrace();
    		}
    	}
    }