RabbitMQ队列的选择及应用场景

// 队列声明
channel.queueDeclare("safe_queue", true, false, false, null); // Durability=true
// 消息发布
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
    .deliveryMode(2) // 持久化消息
    .build();

@Configuration
public class QuorumConfig {
    @Bean
    public Queue quorumQueue() {
        Map<String,Object> params = new HashMap<>();
        params.put("x-queue-type","quorum");
        return new Queue(MyConstants.QUEUE_QUORUM,true,false,false,params);
    }
}

@Configuration
public class StreamConfig {
    @Bean
    public Queue streamQueue() {
        Map<String,Object> params = new HashMap<>();
        params.put("x-queue-type","stream");
        params.put("x-max-length-bytes", 20_000_000_000L); // 指定队列的大小
        params.put("x-stream-max-segment-size-bytes", 100_000_000); // 文件分片存储，每一片的大小
			 //必须设置持久化为true，同时独占和自动删除模式为false
        return new Queue(MyConstants.QUEUE_STREAM,true,false,false,params);
    }
}

public void stremReceiver(Channel channel,String message){
		try {
			channel.basicQos(100);
			Consumer myconsumer = new DefaultConsumer(channel) {
				@Override
				public void handleDelivery(String consumerTag, Envelope envelope,
										   AMQP.BasicProperties properties, byte[] body)
						throws IOException {
					System.out.println("========================");
					String routingKey = envelope.getRoutingKey();
					System.out.println("routingKey >"+routingKey);
					String contentType = properties.getContentType();
					System.out.println("contentType >"+contentType);
					long deliveryTag = envelope.getDeliveryTag();
					System.out.println("deliveryTag >"+deliveryTag);
					System.out.println("content:"+new String(body,"UTF-8"));
					// (process the message components here ...)
					//消息处理完后，进行答复。答复过的消息，服务器就不会再次转发。
					//没有答复过的消息，服务器会一直不停转发。
					channel.basicAck(deliveryTag, false);
				}
			};
			Map<String,Object> consumeParam = new HashMap<>();
			//first: 从日志队列中第一个可消费的消息开始消费
			//last: 消费消息日志中最后一个消息
			//next: 相当于不指定offset，消费不到消息。
			//Offset: 一个数字型的偏
			//Timestamp:一个代表时间的Data类型变量，表示从这个时间点开始消费。
			//例如 一个小时前 Date timestamp = new Date(System.currentTimeMillis() - 60 * 60 * 1_000)
			consumeParam.put("x-stream-offset","last");
			channel.basicConsume(MyConstants.QUEUE_STREAM, false,consumeParam, myconsumer);
		} catch (IOException e) {
			e.printStackTrace();
		}
		System.out.println("quorumReceiver received message : "+ message);
	}

Map<String, Object> args = new HashMap<String, Object>();
args.put("x-queue-mode", "lazy");