处理rabbitmq高可用问题

vhr/mailserver/pom.xml

@@ -22,6 +22,10 @@
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-amqp</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-data-redis</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-mail</artifactId>

vhr/mailserver/src/main/java/org/javaboy/mailserver/receiver/MailReceiver.java

 package org.javaboy.mailserver.receiver;
 
+import com.rabbitmq.client.Channel;
 import org.javaboy.vhr.model.Employee;
 import org.javaboy.vhr.model.MailConstants;
 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.Autowired;
 import org.springframework.boot.autoconfigure.mail.MailProperties;
+import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.mail.javamail.JavaMailSender;
 import org.springframework.mail.javamail.MimeMessageHelper;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageHeaders;
 import org.springframework.stereotype.Component;
 import org.thymeleaf.TemplateEngine;
 import org.thymeleaf.context.Context;
 
 import javax.mail.MessagingException;
 import javax.mail.internet.MimeMessage;
+import java.io.IOException;
 import java.util.Date;
 
 /**
@@ -37,10 +43,21 @@ public class MailReceiver {
     MailProperties mailProperties;
     @Autowired
     TemplateEngine templateEngine;
+    @Autowired
+    StringRedisTemplate redisTemplate;
 
     @RabbitListener(queues = MailConstants.MAIL_QUEUE_NAME)
-    public void handler(Employee employee) {
-        logger.info(employee.toString());
+    public void handler(Message message, Channel channel) throws IOException {
+        Employee employee = (Employee) message.getPayload();
+        MessageHeaders headers = message.getHeaders();
+        Long tag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
+        String msgId = (String) headers.get("spring_returned_message_correlation");
+        if (redisTemplate.opsForHash().entries("mail_log").containsKey(msgId)) {
+            //redis 中包含该 key，说明该消息已经被消费过
+            logger.info(msgId + ":消息已经被消费");
+            channel.basicAck(tag, false);//确认消息已消费
+            return;
+        }
         //收到消息，发送邮件
         MimeMessage msg = javaMailSender.createMimeMessage();
         MimeMessageHelper helper = new MimeMessageHelper(msg);
@@ -57,9 +74,13 @@ public class MailReceiver {
             String mail = templateEngine.process("mail", context);
             helper.setText(mail, true);
             javaMailSender.send(msg);
+            redisTemplate.opsForHash().put("mail_log", msgId, "javaboy");
+            channel.basicAck(tag, false);
+            logger.info(msgId + ":邮件发送成功");
         } catch (MessagingException e) {
+            channel.basicNack(tag, false, true);
             e.printStackTrace();
-            logger.error("邮件发送失败："+e.getMessage());
+            logger.error("邮件发送失败：" + e.getMessage());
         }
     }
 }

vhr/mailserver/src/main/resources/application.properties

 server.port=8082
-
 spring.mail.host=smtp.qq.com
 spring.mail.protocol=smtp
 spring.mail.default-encoding=UTF-8
@@ -8,8 +7,13 @@ spring.mail.username=1510161612@qq.com
 spring.mail.port=587
 spring.mail.properties.mail.stmp.socketFactory.class=javax.net.ssl.SSLSocketFactory
 spring.mail.properties.mail.debug=true
-
 spring.rabbitmq.username=guest
 spring.rabbitmq.password=guest
 spring.rabbitmq.host=192.168.91.128
-spring.rabbitmq.port=5672
\ No newline at end of file
+spring.rabbitmq.port=5672
+spring.rabbitmq.listener.simple.acknowledge-mode=manual
+spring.rabbitmq.listener.simple.prefetch=100
+spring.redis.host=192.168.91.128
+spring.redis.port=6379
+spring.redis.password=123
+spring.redis.database=0
\ No newline at end of file

vhr/vhrserver/vhr-service/src/main/java/org/javaboy/vhr/service/EmployeeService.java

@@ -71,6 +71,7 @@ public class EmployeeService {
             mailSendLog.setTryTime(new Date(System.currentTimeMillis() + 1000 * 60 * MailConstants.MSG_TIMEOUT));
             mailSendLogService.insert(mailSendLog);
             rabbitTemplate.convertAndSend(MailConstants.MAIL_EXCHANGE_NAME, MailConstants.MAIL_ROUTING_KEY_NAME, emp, new CorrelationData(msgId));
+            rabbitTemplate.convertAndSend(MailConstants.MAIL_EXCHANGE_NAME, MailConstants.MAIL_ROUTING_KEY_NAME, emp, new CorrelationData(msgId));
         }
         return result;
     }