mirror of https://github.com/ZhongFuCheng3y/austin
3y
3 years ago
commit
88e88caec1
@ -0,0 +1,37 @@
|
|||||||
|
package com.java3y.austin.handler.receiver.eventbus;
|
||||||
|
|
||||||
|
import com.google.common.eventbus.Subscribe;
|
||||||
|
import com.java3y.austin.common.domain.TaskInfo;
|
||||||
|
import com.java3y.austin.handler.receiver.service.ConsumeService;
|
||||||
|
import com.java3y.austin.support.constans.MessageQueuePipeline;
|
||||||
|
import com.java3y.austin.support.domain.MessageTemplate;
|
||||||
|
import com.java3y.austin.support.mq.eventbus.EventBusListener;
|
||||||
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||||
|
import org.springframework.stereotype.Component;
|
||||||
|
|
||||||
|
import java.util.List;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @author 3y
|
||||||
|
*/
|
||||||
|
@Component
|
||||||
|
@ConditionalOnProperty(name = "austin-mq-pipeline", havingValue = MessageQueuePipeline.EVENT_BUS)
|
||||||
|
public class EventBusReceiver implements EventBusListener {
|
||||||
|
|
||||||
|
@Autowired
|
||||||
|
private ConsumeService consumeService;
|
||||||
|
|
||||||
|
@Override
|
||||||
|
@Subscribe
|
||||||
|
public void consume(List<TaskInfo> lists) {
|
||||||
|
consumeService.consume2Send(lists);
|
||||||
|
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
@Subscribe
|
||||||
|
public void recall(MessageTemplate messageTemplate) {
|
||||||
|
consumeService.consume2recall(messageTemplate);
|
||||||
|
}
|
||||||
|
}
|
||||||
@ -0,0 +1,56 @@
|
|||||||
|
package com.java3y.austin.handler.receiver.service.impl;
|
||||||
|
|
||||||
|
import cn.hutool.core.collection.CollUtil;
|
||||||
|
import com.alibaba.fastjson.JSON;
|
||||||
|
import com.java3y.austin.common.domain.AnchorInfo;
|
||||||
|
import com.java3y.austin.common.domain.LogParam;
|
||||||
|
import com.java3y.austin.common.domain.TaskInfo;
|
||||||
|
import com.java3y.austin.common.enums.AnchorState;
|
||||||
|
import com.java3y.austin.handler.handler.HandlerHolder;
|
||||||
|
import com.java3y.austin.handler.pending.Task;
|
||||||
|
import com.java3y.austin.handler.pending.TaskPendingHolder;
|
||||||
|
import com.java3y.austin.handler.receiver.service.ConsumeService;
|
||||||
|
import com.java3y.austin.handler.utils.GroupIdMappingUtils;
|
||||||
|
import com.java3y.austin.support.domain.MessageTemplate;
|
||||||
|
import com.java3y.austin.support.utils.LogUtils;
|
||||||
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
|
import org.springframework.context.ApplicationContext;
|
||||||
|
import org.springframework.stereotype.Service;
|
||||||
|
|
||||||
|
import java.util.List;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @author 3y
|
||||||
|
*/
|
||||||
|
@Service
|
||||||
|
public class ConsumeServiceImpl implements ConsumeService {
|
||||||
|
private static final String LOG_BIZ_TYPE = "Receiver#consumer";
|
||||||
|
private static final String LOG_BIZ_RECALL_TYPE = "Receiver#recall";
|
||||||
|
@Autowired
|
||||||
|
private ApplicationContext context;
|
||||||
|
|
||||||
|
@Autowired
|
||||||
|
private TaskPendingHolder taskPendingHolder;
|
||||||
|
|
||||||
|
@Autowired
|
||||||
|
private LogUtils logUtils;
|
||||||
|
|
||||||
|
@Autowired
|
||||||
|
private HandlerHolder handlerHolder;
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void consume2Send(List<TaskInfo> taskInfoLists) {
|
||||||
|
String topicGroupId = GroupIdMappingUtils.getGroupIdByTaskInfo(CollUtil.getFirst(taskInfoLists.iterator()));
|
||||||
|
for (TaskInfo taskInfo : taskInfoLists) {
|
||||||
|
logUtils.print(LogParam.builder().bizType(LOG_BIZ_TYPE).object(taskInfo).build(), AnchorInfo.builder().ids(taskInfo.getReceiver()).businessId(taskInfo.getBusinessId()).state(AnchorState.RECEIVE.getCode()).build());
|
||||||
|
Task task = context.getBean(Task.class).setTaskInfo(taskInfo);
|
||||||
|
taskPendingHolder.route(topicGroupId).execute(task);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void consume2recall(MessageTemplate messageTemplate) {
|
||||||
|
logUtils.print(LogParam.builder().bizType(LOG_BIZ_RECALL_TYPE).object(messageTemplate).build());
|
||||||
|
handlerHolder.route(messageTemplate.getSendChannel()).recall(messageTemplate);
|
||||||
|
}
|
||||||
|
}
|
||||||
@ -0,0 +1,14 @@
|
|||||||
|
package com.java3y.austin.support.constans;
|
||||||
|
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 消息队列常量
|
||||||
|
*
|
||||||
|
* @author 3y
|
||||||
|
*/
|
||||||
|
public interface MessageQueuePipeline {
|
||||||
|
String EVENT_BUS = "eventBus";
|
||||||
|
String KAFKA = "kafka";
|
||||||
|
String ROCKET_MQ = "rocketMq";
|
||||||
|
|
||||||
|
}
|
||||||
@ -0,0 +1,27 @@
|
|||||||
|
package com.java3y.austin.support.mq;
|
||||||
|
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @author 3y
|
||||||
|
* 发送数据至消息队列
|
||||||
|
*/
|
||||||
|
public interface SendMqService {
|
||||||
|
/**
|
||||||
|
* 发送消息
|
||||||
|
*
|
||||||
|
* @param topic
|
||||||
|
* @param jsonValue
|
||||||
|
* @param tagId
|
||||||
|
*/
|
||||||
|
void send(String topic, String jsonValue, String tagId);
|
||||||
|
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 发送消息
|
||||||
|
*
|
||||||
|
* @param topic
|
||||||
|
* @param jsonValue
|
||||||
|
*/
|
||||||
|
void send(String topic, String jsonValue);
|
||||||
|
|
||||||
|
}
|
||||||
@ -0,0 +1,27 @@
|
|||||||
|
package com.java3y.austin.support.mq.eventbus;
|
||||||
|
|
||||||
|
|
||||||
|
import com.java3y.austin.common.domain.TaskInfo;
|
||||||
|
import com.java3y.austin.support.domain.MessageTemplate;
|
||||||
|
|
||||||
|
import java.util.List;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @author 3y
|
||||||
|
* 监听器
|
||||||
|
*/
|
||||||
|
public interface EventBusListener {
|
||||||
|
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 消费消息
|
||||||
|
* @param lists
|
||||||
|
*/
|
||||||
|
void consume(List<TaskInfo> lists);
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 撤回消息
|
||||||
|
* @param messageTemplate
|
||||||
|
*/
|
||||||
|
void recall(MessageTemplate messageTemplate);
|
||||||
|
}
|
||||||
@ -0,0 +1,50 @@
|
|||||||
|
package com.java3y.austin.support.mq.kafka;
|
||||||
|
|
||||||
|
import cn.hutool.core.util.StrUtil;
|
||||||
|
import com.java3y.austin.support.constans.MessageQueuePipeline;
|
||||||
|
import com.java3y.austin.support.mq.SendMqService;
|
||||||
|
import lombok.extern.slf4j.Slf4j;
|
||||||
|
import org.apache.kafka.clients.producer.ProducerRecord;
|
||||||
|
import org.apache.kafka.common.header.Header;
|
||||||
|
import org.apache.kafka.common.header.internals.RecordHeader;
|
||||||
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
|
import org.springframework.beans.factory.annotation.Value;
|
||||||
|
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||||
|
import org.springframework.kafka.core.KafkaTemplate;
|
||||||
|
import org.springframework.stereotype.Service;
|
||||||
|
|
||||||
|
import java.nio.charset.StandardCharsets;
|
||||||
|
import java.util.Arrays;
|
||||||
|
import java.util.List;
|
||||||
|
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @author 3y
|
||||||
|
* kafka 发送实现类
|
||||||
|
*/
|
||||||
|
@Slf4j
|
||||||
|
@Service
|
||||||
|
@ConditionalOnProperty(name = "austin-mq-pipeline", havingValue = MessageQueuePipeline.KAFKA)
|
||||||
|
public class KafkaSendMqServiceImpl implements SendMqService {
|
||||||
|
|
||||||
|
@Autowired
|
||||||
|
private KafkaTemplate kafkaTemplate;
|
||||||
|
|
||||||
|
@Value("${austin.business.tagId.key}")
|
||||||
|
private String tagIdKey;
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void send(String topic, String jsonValue, String tagId) {
|
||||||
|
if (StrUtil.isNotBlank(tagId)) {
|
||||||
|
List<Header> headers = Arrays.asList(new RecordHeader(tagIdKey, tagId.getBytes(StandardCharsets.UTF_8)));
|
||||||
|
kafkaTemplate.send(new ProducerRecord(topic, null, null, null, jsonValue, headers));
|
||||||
|
} else {
|
||||||
|
kafkaTemplate.send(topic, jsonValue);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void send(String topic, String jsonValue) {
|
||||||
|
send(topic, jsonValue, null);
|
||||||
|
}
|
||||||
|
}
|
||||||
Loading…
Reference in new issue