mirror of https://github.com/ZhongFuCheng3y/austin
3y
3 years ago
parent
0d6b3ebcea
commit
4264124660
@ -1,40 +0,0 @@
|
||||
package com.java3y.austin;
|
||||
|
||||
import com.alibaba.fastjson.JSON;
|
||||
import com.java3y.austin.domain.TaskInfo;
|
||||
import com.java3y.austin.handler.SmsHandler;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.kafka.clients.consumer.ConsumerRecord;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.kafka.annotation.KafkaListener;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Optional;
|
||||
|
||||
/**
|
||||
* @author 3y
|
||||
* 消费MQ的消息
|
||||
*/
|
||||
@Component
|
||||
@Slf4j
|
||||
public class Receiver {
|
||||
|
||||
@Autowired
|
||||
private SmsHandler smsHandler;
|
||||
|
||||
@KafkaListener(topics = "#{'${austin.topic.name}'}", groupId = "austin")
|
||||
public void consumer(ConsumerRecord<?, String> consumerRecord) {
|
||||
Optional<String> kafkaMessage = Optional.ofNullable(consumerRecord.value());
|
||||
if (kafkaMessage.isPresent()) {
|
||||
List<TaskInfo> lists = JSON.parseArray(kafkaMessage.get(), TaskInfo.class);
|
||||
for (TaskInfo taskInfo : lists) {
|
||||
smsHandler.doHandler(taskInfo);
|
||||
}
|
||||
log.info("receiver message:{}", JSON.toJSONString(lists));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,36 @@
|
||||
package com.java3y.austin.config;
|
||||
|
||||
import com.java3y.austin.pending.Task;
|
||||
import com.java3y.austin.receiver.Receiver;
|
||||
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.context.annotation.Scope;
|
||||
|
||||
/**
|
||||
* Handler模块的配置信息
|
||||
*
|
||||
* @author 3y
|
||||
*/
|
||||
@Configuration
|
||||
public class PrototypeBeanConfig {
|
||||
|
||||
/**
|
||||
* 定义多例的Receiver
|
||||
*/
|
||||
@Bean
|
||||
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
|
||||
public Receiver receiver() {
|
||||
return new Receiver();
|
||||
}
|
||||
|
||||
/**
|
||||
* 定义多例的Task
|
||||
*/
|
||||
@Bean
|
||||
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
|
||||
public Task task() {
|
||||
return new Task();
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,17 @@
|
||||
package com.java3y.austin.handler;
|
||||
|
||||
import com.java3y.austin.domain.TaskInfo;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* 邮件发送处理
|
||||
*
|
||||
* @author 3y
|
||||
*/
|
||||
@Component
|
||||
public class EmailHandler extends Handler {
|
||||
|
||||
@Override
|
||||
public void handler(TaskInfo taskInfoList) {
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,26 @@
|
||||
package com.java3y.austin.handler;
|
||||
|
||||
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* channel->Handler的映射关系
|
||||
*
|
||||
* @author 3y
|
||||
*/
|
||||
@Component
|
||||
public class HandlerHolder {
|
||||
|
||||
private Map<Integer, Handler> handlers = new HashMap<Integer, Handler>(32);
|
||||
|
||||
public void putHandler(Integer channelCode, Handler handler) {
|
||||
handlers.put(channelCode, handler);
|
||||
}
|
||||
|
||||
public Handler route(Integer channelCode) {
|
||||
return handlers.get(channelCode);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,37 @@
|
||||
package com.java3y.austin.pending;
|
||||
|
||||
|
||||
import com.java3y.austin.domain.TaskInfo;
|
||||
import com.java3y.austin.handler.HandlerHolder;
|
||||
import lombok.Data;
|
||||
import lombok.experimental.Accessors;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
|
||||
/**
|
||||
* Task 执行器
|
||||
* 1.通用去重功能
|
||||
* 2.发送消息
|
||||
*
|
||||
* @author 3y
|
||||
*/
|
||||
@Data
|
||||
@Accessors(chain = true)
|
||||
@Slf4j
|
||||
public class Task implements Runnable {
|
||||
|
||||
@Autowired
|
||||
private HandlerHolder handlerHolder;
|
||||
|
||||
private TaskInfo taskInfo;
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
|
||||
// 1. TODO 通用去重
|
||||
|
||||
// 2. 真正发送消息
|
||||
handlerHolder.route(taskInfo.getSendChannel())
|
||||
.doHandler(taskInfo);
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,53 @@
|
||||
package com.java3y.austin.receiver;
|
||||
|
||||
import com.alibaba.fastjson.JSON;
|
||||
import com.java3y.austin.domain.TaskInfo;
|
||||
import com.java3y.austin.pending.Task;
|
||||
import com.java3y.austin.pending.TaskPendingHolder;
|
||||
import com.java3y.austin.utils.GroupIdMappingUtils;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.kafka.clients.consumer.ConsumerRecord;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.context.ApplicationContext;
|
||||
import org.springframework.kafka.annotation.KafkaListener;
|
||||
import org.springframework.kafka.support.KafkaHeaders;
|
||||
import org.springframework.messaging.handler.annotation.Header;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Optional;
|
||||
|
||||
/**
|
||||
* @author 3y
|
||||
* 消费MQ的消息
|
||||
*/
|
||||
@Slf4j
|
||||
public class Receiver {
|
||||
|
||||
@Autowired
|
||||
private ApplicationContext context;
|
||||
|
||||
@Autowired
|
||||
private TaskPendingHolder taskPendingHolder;
|
||||
|
||||
@KafkaListener(topics = "#{'${austin.topic.name}'}")
|
||||
public void consumer(ConsumerRecord<?, String> consumerRecord, @Header(KafkaHeaders.GROUP_ID) String topicGroupId) {
|
||||
Optional<String> kafkaMessage = Optional.ofNullable(consumerRecord.value());
|
||||
if (kafkaMessage.isPresent()) {
|
||||
List<TaskInfo> TaskInfoLists = JSON.parseArray(kafkaMessage.get(), TaskInfo.class);
|
||||
String messageGroupId = GroupIdMappingUtils.getGroupIdByTaskInfo(TaskInfoLists.get(0));
|
||||
|
||||
/**
|
||||
* 每个消费者组 只消费 他们自身关心的消息
|
||||
*/
|
||||
if (topicGroupId.equals(messageGroupId)) {
|
||||
for (TaskInfo taskInfo : TaskInfoLists) {
|
||||
Task task = context.getBean(Task.class).setTaskInfo(taskInfo);
|
||||
taskPendingHolder.route(topicGroupId).execute(task);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
@ -0,0 +1,66 @@
|
||||
package com.java3y.austin.receiver;
|
||||
|
||||
import com.java3y.austin.utils.GroupIdMappingUtils;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.context.ApplicationContext;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.kafka.annotation.KafkaListenerAnnotationBeanPostProcessor;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import javax.annotation.PostConstruct;
|
||||
import java.lang.reflect.Method;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* 启动消费者
|
||||
*
|
||||
* @author 3y
|
||||
* @date 2021/12/4
|
||||
*/
|
||||
@Service
|
||||
public class ReceiverStart {
|
||||
@Autowired
|
||||
private ApplicationContext context;
|
||||
|
||||
/**
|
||||
* receiver的消费方法常量
|
||||
*/
|
||||
private static final String RECEIVER_METHOD_NAME = "Receiver.consumer";
|
||||
|
||||
/**
|
||||
* 获取得到所有的groupId
|
||||
*/
|
||||
private static List<String> groupIds = GroupIdMappingUtils.getAllGroupIds();
|
||||
|
||||
/**
|
||||
* 下标(用于迭代groupIds位置)
|
||||
*/
|
||||
private static Integer index = 0;
|
||||
|
||||
/**
|
||||
* 为每个渠道不同的消息类型 创建一个Receiver对象
|
||||
*/
|
||||
@PostConstruct
|
||||
public void init() {
|
||||
for (int i = 0; i < groupIds.size(); i++) {
|
||||
context.getBean(Receiver.class);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 给每个Receiver对象的consumer方法 @KafkaListener赋值相应的groupId
|
||||
*/
|
||||
@Bean
|
||||
public static KafkaListenerAnnotationBeanPostProcessor.AnnotationEnhancer groupIdEnhancer() {
|
||||
return (attrs, element) -> {
|
||||
if (element instanceof Method) {
|
||||
String name = ((Method) element).getDeclaringClass().getSimpleName() + "." + ((Method) element).getName();
|
||||
if (RECEIVER_METHOD_NAME.equals(name)) {
|
||||
attrs.put("groupId", groupIds.get(index));
|
||||
index++;
|
||||
}
|
||||
}
|
||||
return attrs;
|
||||
};
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,43 @@
|
||||
package com.java3y.austin.utils;
|
||||
|
||||
|
||||
import com.java3y.austin.domain.TaskInfo;
|
||||
import com.java3y.austin.enums.ChannelType;
|
||||
import com.java3y.austin.enums.MessageType;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* groupId 标识着每一个消费者组
|
||||
*
|
||||
* @author 3y
|
||||
*/
|
||||
public class GroupIdMappingUtils {
|
||||
|
||||
/**
|
||||
* 获取所有的groupIds
|
||||
* (不同的渠道不同的消息类型拥有自己的groupId)
|
||||
*/
|
||||
public static List<String> getAllGroupIds() {
|
||||
List<String> groupIds = new ArrayList<>();
|
||||
for (ChannelType channelType : ChannelType.values()) {
|
||||
for (MessageType messageType : MessageType.values()) {
|
||||
groupIds.add(channelType.getCode_en() + "." + messageType.getCode_en());
|
||||
}
|
||||
}
|
||||
return groupIds;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* 根据TaskInfo获取当前消息的groupId
|
||||
* @param taskInfo
|
||||
* @return
|
||||
*/
|
||||
public static String getGroupIdByTaskInfo(TaskInfo taskInfo) {
|
||||
String channelCodeEn = ChannelType.getEnumByCode(taskInfo.getSendChannel()).getCode_en();
|
||||
String msgCodeEn = MessageType.getEnumByCode(taskInfo.getMsgType()).getCode_en();
|
||||
return channelCodeEn + "." + msgCodeEn;
|
||||
}
|
||||
}
|
||||
Loading…
Reference in new issue