springboot+kafka+郵件傳送（最佳實踐）
阿新 • • 發佈：2019-08-01
- 導讀
- 整合spring-kafka，生產者生產郵件message，消費者負責傳送
- 引入執行緒池，多執行緒傳送訊息
- 多郵件伺服器配置
- 定時任務生產訊息；計劃郵件傳送
- 實現過程
- 匯入依賴
<properties>
<java.version>1.8</java.version>
<mysql.version>5.1.38</mysql.version>
<mapper.version>2.1.5</mapper.version>
<mybatis.version>1.3.2</mybatis.version>
<gson.version>2.8.2</gson.version>
<lang3.version>3.4</lang3.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<!-- Spring Test -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-jdbc</artifactId>
</dependency>
<dependency>
<groupId>org.mybatis.spring.boot</groupId>
<artifactId>mybatis-spring-boot-starter</artifactId>
<version>${mybatis.version}</version>
</dependency>
<!--資料庫驅動-->
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>${mysql.version}</version>
</dependency>
<!-- 通用Mapper啟動器 -->
<dependency>
<groupId>tk.mybatis</groupId>
<artifactId>mapper-spring-boot-starter</artifactId>
<version>${mapper.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
<!-- 自定義配置檔案需要 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-configuration-processor</artifactId>
<optional>true</optional>
</dependency>
<!-- 使用SLF4J + Logback 作為日誌框架 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-logging</artifactId>
</dependency>
<!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-mail-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-mail</artifactId>
</dependency>
<dependency>
<groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId>
<version>${gson.version}</version>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
<version>${lang3.version}</version>
</dependency>
</dependencies>2.application.yml配置kafka、郵箱、資料庫引數
本文采用的yml配置方式，若使用的properties的可使用https://www.toyaml.com/index.html這個線上工具轉換；
郵箱配置，若使用的163、或者qq等，自己百度怎麼申請授權碼（一大堆教程）；
資料庫採用的Hikari連線池，號稱java平臺最快的；
# 配置Kafka叢集IP地址,多個IP以逗號隔開:
spring:
kafka:
bootstrap-servers: 你的kafkaIP:埠號
producer:
retries: 2 #傳送失敗後的重複傳送次數key-serializer: org.apache.kafka.common.serialization.StringSerializer #key序列化方式
value-serializer: org.apache.kafka.common.serialization.StringSerializer #value序列化方式
compression-type: gzip #壓縮格式
batch-size: 16384 #批量傳送的訊息數量
buffer-memory: 33554432 #32M的批處理緩衝區
consumer:
auto-offset-reset: earliest #最早未被消費的offsetenable-auto-commit: false #是否開啟自動提交
#auto-commit-interval: 1000 #自動提交的時間間隔
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer #key解碼方式
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer #value解碼方式
group-id: kafka.consumer.group.id.1
max-poll-records: 50
properties:
session-timeout-ms: 20000 #連線超時時間
max-poll-interval-ms: 15000 #手動提交設定與poll的心跳數,如果訊息佇列中沒有訊息，等待毫秒後，呼叫poll()方法。如果佇列中有訊息，立即消費訊息，每次消費的訊息的多少可以通過max.poll.records配置。
max-partition-fetch-bytes: 15728640 #設定拉取資料的大小 15M
client-id: kafkacli
listener:
ack-mode: manual_immediate
datasource:
type: com.zaxxer.hikari.HikariDataSource
driver-class-name: com.mysql.jdbc.Driver
url: * #自己的
username: * #賬號
password: * #密碼
hikari:
minimum-idle: 5
# 空閒連線存活最大時間，預設600000（10分鐘）
idle-timeout: 180000
# 連線池最大連線數，預設是10
maximum-pool-size: 10
# 此屬性控制從池返回的連線的預設自動提交行為,預設值：true
auto-commit: true
# 連線池名稱
pool-name: MyHikariCP
# 此屬性控制池中連線的最長生命週期，值0表示無限生命週期，預設1800000即30分鐘
max-lifetime: 1800000
# 資料庫連線超時時間,預設30秒，即30000
connection-timeout: 30000
connection-test-query: SELECT 1
# 郵箱伺服器配置，以163郵箱為例
mail:
host: smtp.163.com #郵箱伺服器地址
port: 25 #埠
username: * #使用者名稱
password: * #授權密碼
default-encoding: UTF-8
properties:
from: * #使用者名稱
mail:
smtp:
connectiontimeout: 5000
timeout: 3000
writetimeout: 5000
# 郵件模板
thymeleaf:
cache: false
prefix: classpath:/views/
# 郵件附件
servlet:
multipart:
max-file-size: 10MB #限制單個檔案大小
max-request-size: 50MB #限制請求總量
logging:
level:
com.example: debug
pattern:
# console: %d{yyyy/MM/dd-HH:mm:ss} [%thread] %-5level %logger- %msg%n
# file: %d{yyyy/MM/dd-HH:mm} [%thread] %-5level %logger- %msg%n
path: C:\log
# 郵件失敗重試次數
com:
example:
mail:
sendNumber: 3 #郵件傳送失敗重試次數
threadKillTime: 60 #執行緒超時殺死
mybatis:
type-aliases-package: com.example.mail.entity
configuration:
map-underscore-to-camel-case: true
mapper-locations: mappers/*Mapper.xml
# 非同步執行緒配置,配置核心執行緒數
async:
executor:
thread:
core_pool_size: 15 #核心執行緒數量，執行緒池建立時候初始化的執行緒數
max_pool_size: 15 #最大執行緒數，只有在緩衝佇列滿了之後才會申請超過核心執行緒數的執行緒
queue_capacity: 99999 #緩衝佇列，用來緩衝執行任務的佇列
keep_alive_seconds: 60 #當超過了核心執行緒出之外的執行緒在空閒時間到達之後會被銷燬
await_termination_seconds: 30 #設定執行緒池中任務的等待時間，如果超過這個時候還沒有銷燬就強制銷燬，以確保應用最後能夠被關閉，而不是阻塞住。
name:
prefix: async-service-
prefixson: async-service-son
3.郵件訊息格式
{
"mailUid": "郵件唯一標識",
"fromName": "發件人別名",
"fromMail": "發件人地址",
"toMail": "收件人地址（多個郵箱則用逗號","隔開）",
"ccMail": "抄送人地址（多個郵箱則用逗號","隔開）",
"bccMail": "密送人地址（多個郵箱則用逗號","隔開）",
"planSendTime": "計劃郵件時間",
"mailSubject": "郵件主題",
"mailContent": "郵件正文",
"sendNum": 傳送次數,
"serverFlag": "郵件伺服器標識(多郵件伺服器用)"
}
4.kafka生產者、消費者、mail傳送類等主要方法程式碼
//生產者
public void sendToKafkaStandardMessageAsync(MailDTO mailDTO) {
producer = new KafkaProducer<String, Object>(kafkaConfig.producerConfigs());
producer.send(new ProducerRecord<String, Object>(topicName, gson.toJson(mailDTO)), new Callback() {
@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
if (metadata != null) {
log.info("生產訊息成功{},傳送次數{},checksum：{},offset:{},partition:{},topic:{}", mailDTO.getMailUid(),mailDTO.getSendNum(),metadata.checksum(), metadata.offset(), metadata.partition(), metadata.topic());
}
if (exception != null) {
log.info("生產訊息失敗{}", exception.getMessage());
}
}
});
producer.close();
}
//消費者
/**
* 監聽一個Kafka 主題
**/
@KafkaListener(topics = MQConstants.Topic.ITEM_EXCHANGE_NAME)
public void receiveMessageFromKafka(ConsumerRecord<?, ?> record, Acknowledgment ack) {
log.info("監聽訊息,MailUid:{}", gson.fromJson(String.valueOf(record.value()), MailDTO.class).getMailUid());
Optional<?> kafkaMessage = Optional.ofNullable(record.value());
if (kafkaMessage.isPresent()) {
sendMessageService.sendMessages(gson.fromJson(String.valueOf(record.value()), MailDTO.class));
}
ack.acknowledge();//手動提交偏移量
}
//構建複雜郵件資訊類
public void sendMimeMail(MailVo mailVo) {
try {
MimeMessageHelper messageHelper = new MimeMessageHelper(mailSender.createMimeMessage(), true);//true表示支援複雜型別
mailVo.setFrom("這裡讀取配置檔案中配的from地址");//郵件發信人從配置項讀取
messageHelper.setFrom(mailVo.getFrom());//郵件發信人
messageHelper.setSentDate(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse("2019-07-18 12:45:48"));
messageHelper.setTo(mailVo.getTo().split(","));//郵件收信人
messageHelper.setSubject(mailVo.getSubject());//郵件主題
messageHelper.setText(mailVo.getText());//郵件內容
if (!StringUtils.isEmpty(mailVo.getCc())) {//抄送
messageHelper.setCc(mailVo.getCc().split(","));
}
if (!StringUtils.isEmpty(mailVo.getBcc())) {//密送
messageHelper.setCc(mailVo.getBcc().split(","));
}
if (mailVo.getMultipartFiles() != null) {//新增郵件附件
for (MultipartFile multipartFile : mailVo.getMultipartFiles()) {
messageHelper.addAttachment(multipartFile.getOriginalFilename(), multipartFile);
}
}
if (StringUtils.isEmpty((CharSequence) mailVo.getSentDate())) {//傳送時間
mailVo.setSentDate(new Date());
messageHelper.setSentDate(mailVo.getSentDate());
}
mailSender.send(messageHelper.getMimeMessage());//正式傳送郵件
mailVo.setStatus("ok");
log.info("傳送郵件成功：{}->{}", mailVo.getFrom(), mailVo.getTo());
} catch (Exception e) {
throw new RuntimeException(e);//傳送失敗
}
}
5.思路解析，畫圖吧，口述太費勁
完整程式碼： https://github.com/wwt729/mail.git
&n