Skip to content

L
log-platform
Commit 1d77a20e authored by 赵啸非's avatar 赵啸非
Browse files
Options

基础model类提交

parent 7aed0e36
Expand all Show whitespace changes
Inline Side-by-side
Showing with 12201 additions and 10168 deletions
common-lib/src/main/java/com/mortals/xhx/common/model/DefaultTbQueueMsgHeaders.java deleted 100644 → 0
View file @ 7aed0e36
package com.mortals.xhx.common.model;
import cn.hutool.crypto.SecureUtil;
import cn.hutool.crypto.asymmetric.SignAlgorithm;
import com.mortals.framework.util.DateUtils;
import com.mortals.xhx.common.code.MessageProtocolEnum;
import com.mortals.xhx.queue.TbQueueMsgHeaders;
import lombok.Setter;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
/**
* 默认消息头
*
* @author: zxfei
* @date: 2021/11/22 11:14
*/
public class DefaultTbQueueMsgHeaders implements TbQueueMsgHeaders {
@Setter
protected Map<String, String> data = new HashMap<>();
public DefaultTbQueueMsgHeaders() {
data.put(MessageHeader.TIMESTAMP, String.valueOf(new Date().getTime()));
// data.put(MessageHeader.MESSAGESIGN, new String(SecureUtil.sign(SignAlgorithm.SHA256withRSA).sign(data.get(MessageHeader.TIMESTAMP).getBytes())));
// TODO: 2022/4/15
data.put(MessageHeader.MESSAGESIGN, "abcd1234");
data.put(MessageHeader.MESSAGEPROTOCOL, MessageProtocolEnum.JSON.getValue());
// data.put(MessageHeader.TOPIC, "");
// data.put(MessageHeader.QOS, "0");
}
@Override
public void put(String key, String value) {
data.put(key, value);
}
@Override
public String get(String key) {
return data.get(key);
}
@Override
public Map<String, String> getData() {
return data;
}
}
common-lib/src/main/java/com/mortals/xhx/feign/access/IAccessLogFeign.java 0 → 100644
View file @ 1d77a20e
package com.mortals.xhx.feign.access;
import com.mortals.xhx.common.pdu.RespData;
import com.mortals.xhx.common.pdu.access.AccessLogPdu;
import com.alibaba.fastjson.JSON;
import com.mortals.framework.common.Rest;
import com.mortals.xhx.feign.IFeign;
import feign.hystrix.FallbackFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.*;
import java.util.List;
/**
* 访问日志 Feign接口
* @author zxfei
* @date 2022-08-17
*/
@FeignClient(name = "log-platform", path = "/logservice", fallbackFactory = AccessLogFeignFallbackFactory.class)
public interface IAccessLogFeign extends IFeign {
/**
* 查看访问日志列表
*
* @param accessLogPdu
* @return
*/
@PostMapping(value = "/access/log/list")
Rest<RespData<List<AccessLogPdu>>> list(@RequestBody AccessLogPdu accessLogPdu);
/**
* 查看访问日志
*
* @param id
* @return
*/
@GetMapping(value = "/access/log/info")
Rest<AccessLogPdu> info(@RequestParam(value = "id") Long id);
/**
* 删除访问日志
*
* @param ids
* @return
*/
@GetMapping(value = "/access/log/delete")
Rest<Void> delete(Long[] ids,@RequestHeader("Authorization") String authorization);
/**
* 访问日志保存更新
*
* @param accessLogPdu
* @return
*/
@PostMapping(value = "/access/log/save")
Rest<RespData<AccessLogPdu>> save(@RequestBody AccessLogPdu accessLogPdu,@RequestHeader("Authorization") String authorization);
}
@Slf4j
@Component
class AccessLogFeignFallbackFactory implements FallbackFactory<IAccessLogFeign> {
@Override
public IAccessLogFeign create(Throwable t) {
return new IAccessLogFeign() {
@Override
public Rest<RespData<List<AccessLogPdu>>> list(AccessLogPdu accessLogPdu) {
return Rest.fail("暂时无法获取访问日志列表，请稍后再试!");
}
@Override
public Rest<AccessLogPdu> info(Long id) {
return Rest.fail("暂时无法获取访问日志详细，请稍后再试!");
}
@Override
public Rest<Void> delete(Long[] ids, String authorization) {
return Rest.fail("暂时无法删除访问日志，请稍后再试!");
}
@Override
public Rest<RespData<AccessLogPdu>> save(AccessLogPdu accessLogPdu, String authorization) {
return Rest.fail("暂时无法保存访问日志，请稍后再试!");
}
};
}
}
common-lib/src/main/java/com/mortals/xhx/feign/biz/IBizLogFeign.java 0 → 100644
View file @ 1d77a20e
package com.mortals.xhx.feign.biz;
import com.mortals.xhx.common.pdu.RespData;
import com.mortals.xhx.common.pdu.biz.BizLogPdu;
import com.alibaba.fastjson.JSON;
import com.mortals.framework.common.Rest;
import com.mortals.xhx.feign.IFeign;
import feign.hystrix.FallbackFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.*;
import java.util.List;
/**
* 业务日志 Feign接口
* @author zxfei
* @date 2022-08-17
*/
@FeignClient(name = "log-platform", path = "/logservice", fallbackFactory = BizLogFeignFallbackFactory.class)
public interface IBizLogFeign extends IFeign {
/**
* 查看业务日志列表
*
* @param bizLogPdu
* @return
*/
@PostMapping(value = "/biz/log/list")
Rest<RespData<List<BizLogPdu>>> list(@RequestBody BizLogPdu bizLogPdu);
/**
* 查看业务日志
*
* @param id
* @return
*/
@GetMapping(value = "/biz/log/info")
Rest<BizLogPdu> info(@RequestParam(value = "id") Long id);
/**
* 删除业务日志
*
* @param ids
* @return
*/
@GetMapping(value = "/biz/log/delete")
Rest<Void> delete(Long[] ids,@RequestHeader("Authorization") String authorization);
/**
* 业务日志保存更新
*
* @param bizLogPdu
* @return
*/
@PostMapping(value = "/biz/log/save")
Rest<RespData<BizLogPdu>> save(@RequestBody BizLogPdu bizLogPdu,@RequestHeader("Authorization") String authorization);
}
@Slf4j
@Component
class BizLogFeignFallbackFactory implements FallbackFactory<IBizLogFeign> {
@Override
public IBizLogFeign create(Throwable t) {
return new IBizLogFeign() {
@Override
public Rest<RespData<List<BizLogPdu>>> list(BizLogPdu bizLogPdu) {
return Rest.fail("暂时无法获取业务日志列表，请稍后再试!");
}
@Override
public Rest<BizLogPdu> info(Long id) {
return Rest.fail("暂时无法获取业务日志详细，请稍后再试!");
}
@Override
public Rest<Void> delete(Long[] ids, String authorization) {
return Rest.fail("暂时无法删除业务日志，请稍后再试!");
}
@Override
public Rest<RespData<BizLogPdu>> save(BizLogPdu bizLogPdu, String authorization) {
return Rest.fail("暂时无法保存业务日志，请稍后再试!");
}
};
}
}
common-lib/src/main/java/com/mortals/xhx/feign/error/IErrorLogFeign.java 0 → 100644
View file @ 1d77a20e
package com.mortals.xhx.feign.error;
import com.mortals.xhx.common.pdu.RespData;
import com.mortals.xhx.common.pdu.error.ErrorLogPdu;
import com.alibaba.fastjson.JSON;
import com.mortals.framework.common.Rest;
import com.mortals.xhx.feign.IFeign;
import feign.hystrix.FallbackFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.*;
import java.util.List;
/**
* 异常日志 Feign接口
* @author zxfei
* @date 2022-08-17
*/
@FeignClient(name = "log-platform", path = "/logservice", fallbackFactory = ErrorLogFeignFallbackFactory.class)
public interface IErrorLogFeign extends IFeign {
/**
* 查看异常日志列表
*
* @param errorLogPdu
* @return
*/
@PostMapping(value = "/error/log/list")
Rest<RespData<List<ErrorLogPdu>>> list(@RequestBody ErrorLogPdu errorLogPdu);
/**
* 查看异常日志
*
* @param id
* @return
*/
@GetMapping(value = "/error/log/info")
Rest<ErrorLogPdu> info(@RequestParam(value = "id") Long id);
/**
* 删除异常日志
*
* @param ids
* @return
*/
@GetMapping(value = "/error/log/delete")
Rest<Void> delete(Long[] ids,@RequestHeader("Authorization") String authorization);
/**
* 异常日志保存更新
*
* @param errorLogPdu
* @return
*/
@PostMapping(value = "/error/log/save")
Rest<RespData<ErrorLogPdu>> save(@RequestBody ErrorLogPdu errorLogPdu,@RequestHeader("Authorization") String authorization);
}
@Slf4j
@Component
class ErrorLogFeignFallbackFactory implements FallbackFactory<IErrorLogFeign> {
@Override
public IErrorLogFeign create(Throwable t) {
return new IErrorLogFeign() {
@Override
public Rest<RespData<List<ErrorLogPdu>>> list(ErrorLogPdu errorLogPdu) {
return Rest.fail("暂时无法获取异常日志列表，请稍后再试!");
}
@Override
public Rest<ErrorLogPdu> info(Long id) {
return Rest.fail("暂时无法获取异常日志详细，请稍后再试!");
}
@Override
public Rest<Void> delete(Long[] ids, String authorization) {
return Rest.fail("暂时无法删除异常日志，请稍后再试!");
}
@Override
public Rest<RespData<ErrorLogPdu>> save(ErrorLogPdu errorLogPdu, String authorization) {
return Rest.fail("暂时无法保存异常日志，请稍后再试!");
}
};
}
}
common-lib/src/main/java/com/mortals/xhx/feign/operate/IOperateLogFeign.java 0 → 100644
View file @ 1d77a20e
package com.mortals.xhx.feign.operate;
import com.mortals.xhx.common.pdu.RespData;
import com.mortals.xhx.common.pdu.operate.OperateLogPdu;
import com.alibaba.fastjson.JSON;
import com.mortals.framework.common.Rest;
import com.mortals.xhx.feign.IFeign;
import feign.hystrix.FallbackFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.openfeign.FeignClient;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.*;
import java.util.List;
/**
* 平台用户操作日志业务 Feign接口
* @author zxfei
* @date 2022-08-17
*/
@FeignClient(name = "log-platform", path = "/logservice", fallbackFactory = OperateLogFeignFallbackFactory.class)
public interface IOperateLogFeign extends IFeign {
/**
* 查看平台用户操作日志业务列表
*
* @param operateLogPdu
* @return
*/
@PostMapping(value = "/operate/log/list")
Rest<RespData<List<OperateLogPdu>>> list(@RequestBody OperateLogPdu operateLogPdu);
/**
* 查看平台用户操作日志业务
*
* @param id
* @return
*/
@GetMapping(value = "/operate/log/info")
Rest<OperateLogPdu> info(@RequestParam(value = "id") Long id);
/**
* 删除平台用户操作日志业务
*
* @param ids
* @return
*/
@GetMapping(value = "/operate/log/delete")
Rest<Void> delete(Long[] ids,@RequestHeader("Authorization") String authorization);
/**
* 平台用户操作日志业务保存更新
*
* @param operateLogPdu
* @return
*/
@PostMapping(value = "/operate/log/save")
Rest<RespData<OperateLogPdu>> save(@RequestBody OperateLogPdu operateLogPdu,@RequestHeader("Authorization") String authorization);
}
@Slf4j
@Component
class OperateLogFeignFallbackFactory implements FallbackFactory<IOperateLogFeign> {
@Override
public IOperateLogFeign create(Throwable t) {
return new IOperateLogFeign() {
@Override
public Rest<RespData<List<OperateLogPdu>>> list(OperateLogPdu operateLogPdu) {
return Rest.fail("暂时无法获取平台用户操作日志业务列表，请稍后再试!");
}
@Override
public Rest<OperateLogPdu> info(Long id) {
return Rest.fail("暂时无法获取平台用户操作日志业务详细，请稍后再试!");
}
@Override
public Rest<Void> delete(Long[] ids, String authorization) {
return Rest.fail("暂时无法删除平台用户操作日志业务，请稍后再试!");
}
@Override
public Rest<RespData<OperateLogPdu>> save(OperateLogPdu operateLogPdu, String authorization) {
return Rest.fail("暂时无法保存平台用户操作日志业务，请稍后再试!");
}
};
}
}
common-lib/src/main/java/com/mortals/xhx/queue/ConsumerService.java deleted 100644 → 0
View file @ 7aed0e36
package com.mortals.xhx.queue;
import com.alibaba.fastjson.JSONObject;
import lombok.Getter;
import lombok.extern.apachecommons.CommonsLog;
import org.slf4j.Logger;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* 服务端消费消息服务
*
* @author: zxfei
* @date: 2021/11/22 11:31
*/
@CommonsLog
public class ConsumerService {
private long pollDuration;
protected volatile ExecutorService consumersExecutor;
@Getter
private TbQueueConsumer<TbQueueMsg> mainConsumer;
private String currentIp;
protected volatile boolean stopped = false;
public void init(TbQueueConsumer<TbQueueMsg> mainConsumer) {
this.consumersExecutor = Executors.newCachedThreadPool();
this.mainConsumer = mainConsumer;
launchMainConsumers();
this.mainConsumer.subscribe();
}
public ConsumerService(long pollDuration, String currentIp) {
this.pollDuration = pollDuration;
this.currentIp = currentIp;
}
/**
* 消费服务主线程
*/
protected void launchMainConsumers() {
consumersExecutor.submit(() -> {
while (!stopped) {
try {
//todo
List<TbQueueMsg> poll = mainConsumer.poll(pollDuration);
List<TbQueueMsg> msgs = poll;
if (msgs.isEmpty()) {
continue;
}
for (TbQueueMsg item : msgs) {
//todo
// }
}
} catch (Exception e) {
log.error("Exception", e);
}
}
log.info("Queue Consumer stopped.");
});
}
public void destroy() {
if (!stopped) {
stopMainConsumers();
}
stopped = true;
if (consumersExecutor != null) {
consumersExecutor.shutdownNow();
}
}
protected void stopMainConsumers() {
if (mainConsumer != null) {
mainConsumer.unsubscribe();
}
}
}
common-lib/src/main/java/com/mortals/xhx/queue/DefaultTbCoreConsumerService.java deleted 100644 → 0
View file @ 7aed0e36
package com.mortals.xhx.queue;
import com.mortals.xhx.queue.processing.AbstractConsumerService;
import com.mortals.xhx.queue.provider.TbCoreQueueFactory;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationEvent;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
@Service
@Slf4j
public class DefaultTbCoreConsumerService extends AbstractConsumerService<TbQueueMsg> implements TbCoreConsumerService {
@Value("${queue.core.poll-interval}")
private long pollDuration;//队列拉取时间间隔，单位毫秒
@Value("${queue.core.pack-processing-timeout}")
private long packProcessingTimeout;
@Getter
private LinkedBlockingQueue<TbQueueMsg> comsureQueue = new LinkedBlockingQueue<>();
@Getter
private TbQueueConsumer<TbQueueMsg> mainConsumer;
@Getter
private List<TbQueueConsumer<TbQueueMsg>> consumerList;
/**
* 根据配置文件动态加载kafka,rabbmitMq等工厂类
* @param tbCoreQueueFactory
*/
public DefaultTbCoreConsumerService(TbCoreQueueFactory tbCoreQueueFactory) {
//通过工厂类创建通道
this.mainConsumer = tbCoreQueueFactory.createMsgConsumer();
}
@PostConstruct
public void init() {
log.info("初始化消费服务线程");
super.init("core-consumer");
}
@PreDestroy
public void destroy() {
super.destroy();
}
@Override
public void onApplicationEvent(ApplicationEvent applicationEvent) {
}
@Override
protected void launchMainConsumers() {
log.info("启动消费线程！");
consumersExecutor.submit(() -> {
while (!stopped) {
try {
List<TbQueueMsg> msgs = mainConsumer.poll(pollDuration);
if (msgs.isEmpty()) {
continue;
}
for (TbQueueMsg item : msgs) {
comsureQueue.offer(item);
}
mainConsumer.commit();
} catch (Exception e) {
if (!stopped) {
// log.warn("Failed to obtain messages from queue.", e);
try {
Thread.sleep(pollDuration);
} catch (InterruptedException e2) {
// log.trace("Failed to wait until the server has capacity to handle new requests", e2);
}
}
}
}
log.info(" Core Consumer stopped.");
});
}
@Override
protected void stopMainConsumers() {
if (mainConsumer != null) {
mainConsumer.unsubscribe();
}
}
@Override
protected void launchConsumersList() {
log.info("启动消费线程组！");
consumerList.stream().forEach(consumer -> {
log.info("channel number:{}"+consumer.getChannelNumber());
consumersExecutor.submit(() -> {
while (!stopped) {
try {
List<TbQueueMsg> msgs = consumer.poll(pollDuration);
if (msgs.isEmpty()) {
continue;
}
for (TbQueueMsg item : msgs) {
comsureQueue.offer(item);
}
consumer.commit();
} catch (Exception e) {
if (!stopped) {
try {
Thread.sleep(pollDuration);
} catch (InterruptedException e2) {
log.trace("Failed to wait until the server has capacity to handle new requests", e2);
}
}
}
}
});
});
}
}
common-lib/src/main/java/com/mortals/xhx/queue/DefaultTbQueueMsg.java deleted 100644 → 0
View file @ 7aed0e36
package com.mortals.xhx.queue;
import cn.hutool.core.util.IdUtil;
import com.alibaba.fastjson.JSON;
import com.mortals.framework.util.DateUtils;
import com.mortals.xhx.common.model.DefaultTbQueueMsgHeaders;
import com.mortals.xhx.common.model.MessageHeader;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.json.JSONObject;
import java.util.HashMap;
import java.util.UUID;
/**
* 默认消息
*
* @author: zxfei
* @date: 2021/11/22 10:59
*/
@Data
@AllArgsConstructor
@NoArgsConstructor
public class DefaultTbQueueMsg implements TbQueueMsg {
/**
* key 唯一标识
*/
private String key;
/**
* 数据载体
*/
private String data;
/**
* 消息头信息
*/
private TbQueueMsgHeaders headers;
public DefaultTbQueueMsg(TbQueueMsg msg) {
this.key = msg.getKey();
this.data = msg.getData();
TbQueueMsgHeaders headers = new DefaultTbQueueMsgHeaders();
msg.getHeaders().getData().entrySet().stream().forEach(item->
headers.put(item.getKey(),item.getValue()));
this.headers = headers;
}
public static void main(String[] args) {
TbQueueMsgHeaders header = new DefaultTbQueueMsgHeaders();
header.put(MessageHeader.MESSAGETYPE, "UPGREAD");
// header.put(MessageHeader.CLIENTID, "abcd1234");
// header.put(MessageHeader.TIMESTAMP, DateUtils.getCurrStrDateTime());
//header.put(MessageHeader.MESSAGESIGN,"ssdafasdfasdfasfd");
TbQueueMsg queueMsg = new DefaultTbQueueMsg(IdUtil.fastUUID(), "eyJmbG93bnVtIjoiQzEwMTEifQ==" , header);
String ret = JSON.toJSONString(queueMsg);
System.out.println("pro:"+ret);
//
// DefaultTbQueueMsg qu = JSON.parseObject(ret, DefaultTbQueueMsg.class);
}
}
common-lib/src/main/java/com/mortals/xhx/queue/MessageQueueFactory.java deleted 100644 → 0
View file @ 7aed0e36
package com.mortals.xhx.queue;
import com.mortals.xhx.common.BrokerConfig;
/**
* 消息队列工厂类，初始化MQ类型（kafka,rabbitMQ，memory）
*
* @author: zxfei
* @date: 2021/11/22 10:57
*/
public interface MessageQueueFactory {
/**
* 创建消息生产者
* @return
*/
TbQueueProducer<TbQueueMsg> createMsgProducer(BrokerConfig brokerConfig);
/**
* 创建消息消费者
* @return
*/
TbQueueConsumer<TbQueueMsg> createMsgConsumer(BrokerConfig brokerConfig);
}
common-lib/src/main/java/com/mortals/xhx/queue/TbCoreConsumerService.java deleted 100644 → 0
View file @ 7aed0e36
package com.mortals.xhx.queue;
import org.springframework.context.ApplicationListener;
public interface TbCoreConsumerService extends ApplicationListener {
}
common-lib/src/main/java/com/mortals/xhx/queue/TbQueueCallback.java deleted 100644 → 0
View file @ 7aed0e36
package com.mortals.xhx.queue;
/**
* 队列回调消息
*
* @author: zxfei
* @date: 2021/11/22 10:57
*/
public interface TbQueueCallback {
void onSuccess(TbQueueMsgMetadata metadata);
void onFailure(Throwable t);
}
common-lib/src/main/java/com/mortals/xhx/queue/TbQueueConsumer.java deleted 100644 → 0
View file @ 7aed0e36
package com.mortals.xhx.queue;
import java.util.List;
import java.util.Set;
/**
* 队列消息消费者接口
*
* @author: zxfei
* @date: 2021/11/22 10:57
*/
public interface TbQueueConsumer<T extends TbQueueMsg> {
/**
* 获取当topic
* @return
*/
String getTopic();
/**
* 订阅
*/
void subscribe();
/**
* 订阅（分区）
* @param partitions
*/
void subscribe(Set<TopicPartitionInfo> partitions);
/**
* 取消订阅
*/
void unsubscribe();
/**
* 取消订阅消息
* @param partitions
*/
void unsubscribe(Set<TopicPartitionInfo> partitions);
/**
* 拉取消息间隔
* @param durationInMillis
* @return
*/
List<T> poll(long durationInMillis);
/**
* 提交
*/
void commit();
/**
* 通道
* @return
*/
String getChannelNumber();
}
common-lib/src/main/java/com/mortals/xhx/queue/TbQueueCoreSettings.java deleted 100644 → 0
View file @ 7aed0e36
package com.mortals.xhx.queue;
import lombok.Data;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
@Data
@Component
public class TbQueueCoreSettings {
@Value("${queue.core.topic}")
private String topic;
@Value("${queue.core.partitions}")
private int partitions;
}
common-lib/src/main/java/com/mortals/xhx/queue/TbQueueMsg.java deleted 100644 → 0
View file @ 7aed0e36
package com.mortals.xhx.queue;
import java.util.UUID;
/**
* 队列消息体
*
* @author: zxfei
* @date: 2021/11/22 10:56
*/
public interface TbQueueMsg {
String getKey();
TbQueueMsgHeaders getHeaders();
String getData();
}
common-lib/src/main/java/com/mortals/xhx/queue/TbQueueMsgDecoder.java deleted 100644 → 0
View file @ 7aed0e36
package com.mortals.xhx.queue;
public interface TbQueueMsgDecoder<T extends TbQueueMsg> {
T decode(TbQueueMsg msg);
}
common-lib/src/main/java/com/mortals/xhx/queue/TbQueueMsgHeaders.java deleted 100644 → 0
View file @ 7aed0e36
package com.mortals.xhx.queue;
import java.util.Map;
/**
* 消息头信息
*
* @author: zxfei
* @date: 2021/11/22 10:56
*/
public interface TbQueueMsgHeaders {
void put(String key, String value);
String get(String key);
Map<String, String> getData();
void setData(Map<String, String> data);
}
common-lib/src/main/java/com/mortals/xhx/queue/TbQueueMsgMetadata.java deleted 100644 → 0
View file @ 7aed0e36
package com.mortals.xhx.queue;
/**
* 队列消息元数据
*
* @author: zxfei
* @date: 2021/11/22 10:56
*/
public interface TbQueueMsgMetadata {
String getMessageId();
}
common-lib/src/main/java/com/mortals/xhx/queue/TbQueueProducer.java deleted 100644 → 0
View file @ 7aed0e36
package com.mortals.xhx.queue;
/**
* 队列消息生产者
*
* @author: zxfei
* @date: 2021/11/22 10:55
*/
public interface TbQueueProducer<T extends TbQueueMsg> {
void init();
String getDefaultTopic();
//发送消息
void send(TopicPartitionInfo tpi, T msg, TbQueueCallback callback);
void stop();
void queueDeclare(TopicPartitionInfo tpi, TbQueueCallback callback);
void queueDel(String queue, TbQueueCallback callback);
}
common-lib/src/main/java/com/mortals/xhx/queue/TopicPartitionInfo.java deleted 100644 → 0
View file @ 7aed0e36
package com.mortals.xhx.queue;
import lombok.Builder;
import lombok.Data;
import java.util.Objects;
import java.util.Optional;
@Data
public class TopicPartitionInfo {
/**
* topic名称
*/
private String topic;
/**
* 分区，kafka存在
*/
private Integer partition;
/**
* 交换机名称，rabbmitmq存在
*/
private String exchangeName;
/**
* 带分区的topic
*/
private String fullTopicName;
@Builder
public TopicPartitionInfo(String topic, Integer partition, String exchangeName) {
this.topic = topic;
this.partition = partition;
this.exchangeName = exchangeName;
String tmp = topic;
if (partition != null) {
tmp += "." + partition;
}
this.fullTopicName = tmp;
}
public TopicPartitionInfo newByTopic(String topic) {
return new TopicPartitionInfo(topic, this.partition, "");
}
public String getTopic() {
return topic;
}
public Optional<Integer> getPartition() {
return Optional.ofNullable(partition);
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
TopicPartitionInfo that = (TopicPartitionInfo) o;
return topic.equals(that.topic) &&
Objects.equals(partition, that.partition) &&
fullTopicName.equals(that.fullTopicName);
}
@Override
public int hashCode() {
return Objects.hash(fullTopicName);
}
}
common-lib/src/main/java/com/mortals/xhx/queue/kafka/KafkaTbQueueMsgMetadata.java deleted 100644 → 0
View file @ 7aed0e36
package com.mortals.xhx.queue.kafka;
import com.mortals.xhx.queue.TbQueueMsgMetadata;
import lombok.AllArgsConstructor;
import lombok.Data;
import org.apache.kafka.clients.producer.RecordMetadata;
/**
* 队列元数据
*
* @author: zxfei
* @date: 2021/11/22 14:40
*/
@Data
@AllArgsConstructor
public class KafkaTbQueueMsgMetadata implements TbQueueMsgMetadata {
private RecordMetadata metadata;
private String messageId;
}
db/menu.sql
View file @ 1d77a20e
This diff is collapsed.
db/module.sql
View file @ 1d77a20e
This diff is collapsed.
db/params.sql
View file @ 1d77a20e
This diff is collapsed.
doc/api.md deleted 100644 → 0
View file @ 7aed0e36
This diff is collapsed.
doc/后端框架使用说明.txt deleted 100644 → 0
View file @ 7aed0e36
This diff is collapsed.
log-manager/pom.xml
View file @ 1d77a20e
This diff is collapsed.
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment