Map<String,Object> args = new HashMap<String,Object>();

args.put("x-max-priority", 10);

channel.queueDeclare("queue_priority", true, false, false, args);

AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();

builder.priority(5);

AMQP.BasicProperties properties = builder.build();

channel.basicPublish("exchange_priority","rk_priority",properties,("messages").getBytes());

public class Producer {

public static final String ip = "10.0.40.127";

public static final int port = 5672;

public static final String username = "admin";

public static final String password = "123456";

public static void main(String[] args) throws IOException{

ConnectionFactory connectionFactory = new ConnectionFactory();

connectionFactory.setPassword(password);

connectionFactory.setUsername(username);

connectionFactory.setPort(port);

connectionFactory.setHost(ip);

/*   Connection connection = connectionFactory.newConnection();

Channel channel = connection.createChannel();

//create exchange

channel.exchangeDeclare("exchange_priority", "direct", true);

//create queue with priority

Map<String, Object> params = new HashMap<>();

params.put("x-max-priority", 10);

channel.queueDeclare("queue_priority", true, false, false, params);

channel.queueBind("queue_priority", "exchange_priority", "rk_priority");

//send message with priority

for (int i = 0; i < 10; i++) {

AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();

if (i % 2 == 0) {

builder.priority(5);

}

AMQP.BasicProperties properties = builder.build();

channel.basicPublish("exchange_priority", "rk_priority", properties, ("produce messages-" + i).getBytes());

}

channel.close();

connection.close();*/

}

}

public class Consumer {

public static final String ip = "10.0.40.127";

public static final int port = 5672;

public static final String username = "admin";

public static final String password = "123456";

public static void main(String[] args) throws IOException, InterruptedException {

ConnectionFactory connectionFactory = new ConnectionFactory();

connectionFactory.setPassword(password);

connectionFactory.setUsername(username);

connectionFactory.setPort(port);

connectionFactory.setHost(ip);

/*        Connection connection = connectionFactory.newConnection();

Channel channel = connection.createChannel();

QueueingConsumer consumer = new QueueingConsumer(channel);

channel.basicConsume("queue_priority",consumer);

while (true) {

QueueingConsumer.Delivery delivery = consumer.nextDelivery();

String msg = new String(delivery.getBody());

System.out.println(msg);

channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

}*/

}

}

void basicQos(int prefetchSize, int prefetchCount, boolean global) throws IOException;

channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);

erver.port=8889

spring.rabbitmq.host=192.168.221.150

spring.rabbitmq.port=5672

spring.rabbitmq.username=zl

spring.rabbitmq.password=123

#开启消息确认机制

spring.rabbitmq.publisher-confirms=true

#支持消息发送失败返回队列

spring.rabbitmq.publisher-returns=true

#设置为 true 后 消费者在消息没有被路由到合适队列情况下会被return监听，而不会自动删除

spring.rabbitmq.template.mandatory=true

spring.rabbitmq.connection-timeout=15000

#用户虚拟机权限名称

spring.rabbitmq.virtual-host=/

#设置消费端手动 ack   none不确认  auto自动确认  manual手动确认

spring.rabbitmq.listener.simple.acknowledge-mode=manual

#消费者最小数量

spring.rabbitmq.listener.simple.concurrency=1

#消费之最大数量

spring.rabbitmq.listener.simple.max-concurrency=1

#开启消费者重试机制(为false时关闭消费者重试，这时消费端代码异常会一直重复收到消息)

spring.rabbitmq.listener.simple.retry.enabled=true

#重试次数5

spring.rabbitmq.listener.simple.retry.max-attempts=5

#重试时间间隔

spring.rabbitmq.listener.simple.retry.initial-interval=5000

#重试次数超过上面的设置之后是否丢弃（false不丢弃时需要写相应代码将该消息加入死信队列）

spring.rabbitmq.listener.simple.default-requeue-rejected=true

#在单个请求中处理的消息个数，他应该大于等于事务数量(unack的最大数量)

spring.rabbitmq.listener.simple.prefetch=2

MessageRecoverer接口实现类

RejectAndDontRequeueRecoverer

RepublishMessageRecoverer

ImmediateRequeueMessageRecoverer

/**

* 死信交换机

* @return

*/

@Bean

public DirectExchange dlxExchange(){

return new DirectExchange(dlxExchangeName);

}

/**

* 死信队列

* @return

*/

@Bean

public Queue dlxQueue(){

return new Queue(dlxQueueName);

}

/**

* 死信队列绑定死信交换机

* @param dlxQueue

* @param dlxExchange

* @return

*/

@Bean

public Binding dlcBinding(Queue dlxQueue, DirectExchange dlxExchange){

return BindingBuilder.bind(dlxQueue).to(dlxExchange).with(dlxRoutingKey);

}

/**

* 业务队列

* @return

*/

@Bean

public Queue queue(){

Map<String,Object> params = new HashMap<>();

params.put("x-dead-letter-exchange",dlxExchangeName);//声明当前队列绑定的死信交换机

params.put("x-dead-letter-routing-key",dlxRoutingKey);//声明当前队列的死信路由键

return QueueBuilder.durable(queueName).withArguments(params).build();

//return new Queue(queueName,true);

}

问题

解决方案

消息重复消费问题：消费者消息处理了，没来的及提交offset，再重启可能导致重复消费

方式一：使用全局MessageID判断消费方使用同一个，解决幂等性。

方式二：用一个消息消费表来记录每一条消息，给每个一个消息设置一个id(uuid)，消费了就保存到表中去。消息过来的时候先查询是否已经消费。