代码基础怎么写我就不说了，看我的另一篇博客
springBoot整合RabbitMQ（Demo）_我要用代码向我喜欢的女孩表白的博客-CSDN博客
假设你要手动ack，怎么做呢？
通常自动是，mq发给服务端，服务端收到了就当处理过了，但是我们要保证数据不丢失。所以得处理完了，才告诉mq说我做了。所以做完了在ack确认，遇到了异常就重发。比如a,b,c，三台机器发给了b,b收到没有处理完，就被关机了。那这条消息就没有发完。
依赖
<!--MQ--> <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId> </dependency>
application.properties的配置
spring.rabbitmq.host=xxxx spring.rabbitmq.port=5672 spring.rabbitmq.username=xxxxx spring.rabbitmq.password=xxxxx spring.rabbitmq.listener.simple.cknowledge-mode=manual spring.rabbitmq.listener.simple.prefetch=1 spring.rabbitmq.publisher-returns=true
下面黄色这一段就是ack手动确认。
MqReceiveDevListener消费者
package com.dengta.tanzhiwcustomermarket.config;import com.dengta.tanzhiwcustomermarket.tools.Constants;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;import javax.annotation.Resource;
import java.io.IOException;@Component
public class MqReceiveDevListener {//@Value("${dev-gen-audio-url}")
//String genAudioUrl;
//
//@Value("${dev-audio-token}")
//String audioToken;private static final Logger logger = LoggerFactory.getLogger(MqReceiveDevListener.class);//---------------------------下面为dev环境下对应的消息队列-----------------------------------------//todo 1.多个消费者并行消费@RabbitListener(queues = Constants.DT_MARKET_MQ_TASK,concurrency = "3")/*** 这个content是一个json，由barrageController发来的*/public void receiveDouYinBarrageDev(String content, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {try {Thread.sleep(1000*10);} catch (Exception e) {e.printStackTrace();}finally {try {channel.basicAck(tag, false);} catch (IOException e) {e.printStackTrace();}}}}
