LiRuochen_WorkFlow/MaxKey
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

message queue support none, Kafka ,RocketMQ

Browse Source 
none, Kafka ,RocketMQ
This commit is contained in:
MaxKey
2022-02-23 14:54:14 +08:00
parent c9317e4a91
commit 090f4c9cc0
18 changed files with 1121 additions and 962 deletions
Download Patch File Download Diff File Expand all files Collapse all files
maxkey-persistence/src/main/java/org/maxkey/persistence/kafka/KafkaIdentityAction.java → maxkey-persistence/src/main/java/org/maxkey/persistence/mq/MqIdentityAction.java
+2 -2
View File
@@ -15,9 +15,9 @@
*/
package org.maxkey.persistence.kafka;
package org.maxkey.persistence.mq;
public class KafkaIdentityAction {
public class MqIdentityAction {
public static String CREATE_ACTION = "CREATE_ACTION";
maxkey-persistence/src/main/java/org/maxkey/persistence/kafka/KafkaIdentityTopic.java → maxkey-persistence/src/main/java/org/maxkey/persistence/mq/MqIdentityTopic.java
+2 -2
View File
@@ -15,9 +15,9 @@
*/
package org.maxkey.persistence.kafka;
package org.maxkey.persistence.mq;
public class KafkaIdentityTopic {
public class MqIdentityTopic {
public final static String USERINFO_TOPIC = "MXK_IDENTITY_USERINFO_TOPIC";
maxkey-persistence/src/main/java/org/maxkey/persistence/kafka/KafkaMessage.java → maxkey-persistence/src/main/java/org/maxkey/persistence/mq/MqMessage.java
+13 -13
View File
@@ -15,14 +15,14 @@
*/
package org.maxkey.persistence.kafka;
package org.maxkey.persistence.mq;
public class KafkaMessage {
public class MqMessage {
String id;
String topic;
String actionType;
String sendTime;
String msgId;
Object content;
public String getTopic() {
@@ -49,15 +49,15 @@ public class KafkaMessage {
this.sendTime = sendTime;
}
public String getMsgId() {
return msgId;
}
public String getId() {
return id;
}
public void setMsgId(String msgId) {
this.msgId = msgId;
}
public void setId(String id) {
this.id = id;
}
public Object getContent() {
public Object getContent() {
return content;
}
@@ -65,15 +65,15 @@ public class KafkaMessage {
this.content = content;
}
public KafkaMessage() {
public MqMessage() {
}
public KafkaMessage(String topic, String actionType, String sendTime, String msgId, Object content) {
public MqMessage(String id,String topic, String actionType, String sendTime, Object content) {
super();
this.id = id;
this.topic = topic;
this.actionType = actionType;
this.sendTime = sendTime;
this.msgId = msgId;
this.content = content;
}
maxkey-persistence/src/main/java/org/maxkey/persistence/kafka/KafkaPersistService.java → maxkey-persistence/src/main/java/org/maxkey/persistence/mq/MqPersistService.java
+25 -43
View File
@@ -15,12 +15,14 @@
*/
package org.maxkey.persistence.kafka;
package org.maxkey.persistence.mq;
import java.util.UUID;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.maxkey.configuration.ApplicationConfig;
import org.maxkey.pretty.PrettyFactory;
import org.maxkey.persistence.mq.thread.KafkaProvisioningThread;
import org.maxkey.persistence.mq.thread.RocketMQProvisioningThread;
import org.maxkey.util.DateUtils;
import org.maxkey.util.JsonUtils;
import org.slf4j.Logger;
@@ -30,14 +32,17 @@ import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
@Component
public class KafkaPersistService {
private static final Logger _logger = LoggerFactory.getLogger(KafkaPersistService.class);
public class MqPersistService {
private static final Logger _logger = LoggerFactory.getLogger(MqPersistService.class);
@Autowired
protected ApplicationConfig applicationConfig;
@Autowired
protected KafkaTemplate<String, String> kafkaTemplate;
@Autowired
private RocketMQTemplate rocketMQTemplate;
public void setApplicationConfig(ApplicationConfig applicationConfig) {
this.applicationConfig = applicationConfig;
@@ -58,50 +63,27 @@ public class KafkaPersistService {
* @param actionType CREATE UPDATE DELETE
*/
public void send(String topic,Object content,String actionType) {
//maxkey.server.kafka.support , if true
if(applicationConfig.isKafkaSupport()) {
KafkaMessage message =
new KafkaMessage(
topic, //kafka TOPIC
actionType, //action of content
DateUtils.getCurrentDateTimeAsString(), //send to kafka time
//maxkey.server.message.queue , if not none , Kafka , RocketMQ
if(applicationConfig.isMessageQueueSupport()) {
MqMessage message =
new MqMessage(
UUID.randomUUID().toString(), //message id as uuid
topic, //TOPIC
actionType, //action of content
DateUtils.getCurrentDateTimeAsString(), //send time
content //content Object to json message content
);
String msg = JsonUtils.gson2Json(message);
//sand msg to Kafka topic
KafkaProvisioningThread thread =
new KafkaProvisioningThread(kafkaTemplate,topic,msg);
//sand msg to MQ topic
Thread thread = null;
if(applicationConfig.getMessageQueue().equalsIgnoreCase("Kafka")) {
_logger.trace("Kafka message...");
thread = new KafkaProvisioningThread(kafkaTemplate,topic,msg);
}else if(applicationConfig.getMessageQueue().equalsIgnoreCase("RocketMQ")) {
_logger.trace("RocketMQ message...");
thread = new RocketMQProvisioningThread(rocketMQTemplate,topic,msg);
}
thread.start();
}
}
/**
* KafkaProvisioningThread for send message
*
*/
class KafkaProvisioningThread extends Thread{
KafkaTemplate<String, String> kafkaTemplate;
String topic ;
String msg;
public KafkaProvisioningThread(
KafkaTemplate<String, String> kafkaTemplate,
String topic,
String msg) {
this.kafkaTemplate = kafkaTemplate;
this.topic = topic;
this.msg = msg;
}
@Override
public void run() {
_logger.debug("send message \n{}" , PrettyFactory.getJsonPretty().format(msg));
kafkaTemplate.send(topic, msg);
_logger.debug("send to Message Queue finished .");
}
}
}
maxkey-persistence/src/main/java/org/maxkey/persistence/mq/thread/KafkaProvisioningThread.java
+52
View File
@@ -0,0 +1,52 @@
/*
* Copyright [2022] [MaxKey of copyright http://www.maxkey.top]
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.maxkey.persistence.mq.thread;
import org.maxkey.pretty.PrettyFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
/**
* Kafka Provisioning Thread for send message
*
*/
public class KafkaProvisioningThread extends Thread{
private static final Logger _logger = LoggerFactory.getLogger(KafkaProvisioningThread.class);
KafkaTemplate<String, String> kafkaTemplate;
String topic ;
String msg;
public KafkaProvisioningThread(
KafkaTemplate<String, String> kafkaTemplate,
String topic,
String msg) {
this.kafkaTemplate = kafkaTemplate;
this.topic = topic;
this.msg = msg;
}
@Override
public void run() {
_logger.debug("send message \n{}" , PrettyFactory.getJsonPretty().format(msg));
kafkaTemplate.send(topic, msg);
_logger.debug("send to Message Queue finished .");
}
}
maxkey-persistence/src/main/java/org/maxkey/persistence/mq/thread/RocketMQProvisioningThread.java
+54
View File
@@ -0,0 +1,54 @@
/*
* Copyright [2022] [MaxKey of copyright http://www.maxkey.top]
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.maxkey.persistence.mq.thread;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.maxkey.pretty.PrettyFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.support.MessageBuilder;
/**
* RocketMQ Provisioning Thread for send message
*
*/
public class RocketMQProvisioningThread extends Thread{
private static final Logger _logger = LoggerFactory.getLogger(RocketMQProvisioningThread.class);
RocketMQTemplate rocketMQTemplate;
String topic ;
String msg;
public RocketMQProvisioningThread(
RocketMQTemplate rocketMQTemplate,
String topic,
String msg) {
this.rocketMQTemplate = rocketMQTemplate;
this.topic = topic;
this.msg = msg;
}
@Override
public void run() {
_logger.debug("send message \n{}" , PrettyFactory.getJsonPretty().format(msg));
rocketMQTemplate.syncSend(topic, MessageBuilder.withPayload(msg).build());
_logger.debug("send to Message Queue finished .");
}
}
maxkey-persistence/src/main/java/org/maxkey/persistence/service/AccountsService.java
+16 -16
View File
@@ -26,10 +26,10 @@ import org.maxkey.entity.Accounts;
import org.maxkey.entity.AccountsStrategy;
import org.maxkey.entity.OrganizationsCast;
import org.maxkey.entity.UserInfo;
import org.maxkey.persistence.kafka.KafkaIdentityAction;
import org.maxkey.persistence.kafka.KafkaIdentityTopic;
import org.maxkey.persistence.kafka.KafkaPersistService;
import org.maxkey.persistence.mapper.AccountsMapper;
import org.maxkey.persistence.mq.MqIdentityAction;
import org.maxkey.persistence.mq.MqIdentityTopic;
import org.maxkey.persistence.mq.MqPersistService;
import org.maxkey.util.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Repository;
@@ -45,7 +45,7 @@ import net.sourceforge.pinyin4j.format.exception.BadHanyuPinyinOutputFormatCombi
public class AccountsService extends JpaBaseService<Accounts>{
@Autowired
KafkaPersistService kafkaPersistService;
MqPersistService mqPersistService;
@Autowired
UserInfoService userInfoService;
@@ -71,17 +71,17 @@ public class AccountsService extends JpaBaseService<Accounts>{
public boolean insert(Accounts account) {
if (super.insert(account)) {
if(kafkaPersistService.getApplicationConfig().isKafkaSupport()) {
if(mqPersistService.getApplicationConfig().isMessageQueueSupport()) {
UserInfo loadUserInfo = userInfoService.findUserRelated(account.getUserId());
account.setUserInfo(loadUserInfo);
OrganizationsCast cast = new OrganizationsCast();
cast.setProvider(account.getAppId());
cast.setOrgId(loadUserInfo.getDepartmentId());
account.setOrgCast(organizationsCastService.query(cast));
kafkaPersistService.send(
KafkaIdentityTopic.ACCOUNT_TOPIC,
mqPersistService.send(
MqIdentityTopic.ACCOUNT_TOPIC,
account,
KafkaIdentityAction.CREATE_ACTION);
MqIdentityAction.CREATE_ACTION);
}
return true;
@@ -91,17 +91,17 @@ public class AccountsService extends JpaBaseService<Accounts>{
public boolean update(Accounts account) {
if (super.update(account)) {
if(kafkaPersistService.getApplicationConfig().isKafkaSupport()) {
if(mqPersistService.getApplicationConfig().isMessageQueueSupport()) {
UserInfo loadUserInfo = userInfoService.findUserRelated(account.getUserId());
account.setUserInfo(loadUserInfo);
OrganizationsCast cast = new OrganizationsCast();
cast.setProvider(account.getAppId());
cast.setOrgId(loadUserInfo.getDepartmentId());
account.setOrgCast(organizationsCastService.query(cast));
kafkaPersistService.send(
KafkaIdentityTopic.ACCOUNT_TOPIC,
mqPersistService.send(
MqIdentityTopic.ACCOUNT_TOPIC,
account,
KafkaIdentityAction.UPDATE_ACTION);
MqIdentityAction.UPDATE_ACTION);
}
return true;
@@ -113,13 +113,13 @@ public class AccountsService extends JpaBaseService<Accounts>{
Accounts account = this.get(id);
if (super.remove(id)) {
UserInfo loadUserInfo = null;
if(kafkaPersistService.getApplicationConfig().isKafkaSupport()) {
if(mqPersistService.getApplicationConfig().isMessageQueueSupport()) {
loadUserInfo = userInfoService.findUserRelated(account.getUserId());
account.setUserInfo(loadUserInfo);
kafkaPersistService.send(
KafkaIdentityTopic.ACCOUNT_TOPIC,
mqPersistService.send(
MqIdentityTopic.ACCOUNT_TOPIC,
account,
KafkaIdentityAction.DELETE_ACTION);
MqIdentityAction.DELETE_ACTION);
}
return true;
maxkey-persistence/src/main/java/org/maxkey/persistence/service/OrganizationsService.java
+10 -10
View File
@@ -23,10 +23,10 @@ import org.apache.mybatis.jpa.persistence.JpaBaseService;
import org.apache.poi.ss.usermodel.Cell;
import org.apache.poi.ss.usermodel.CellType;
import org.maxkey.entity.Organizations;
import org.maxkey.persistence.kafka.KafkaIdentityAction;
import org.maxkey.persistence.kafka.KafkaIdentityTopic;
import org.maxkey.persistence.kafka.KafkaPersistService;
import org.maxkey.persistence.mapper.OrganizationsMapper;
import org.maxkey.persistence.mq.MqIdentityAction;
import org.maxkey.persistence.mq.MqIdentityTopic;
import org.maxkey.persistence.mq.MqPersistService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Repository;
@@ -35,7 +35,7 @@ import org.springframework.stereotype.Repository;
public class OrganizationsService extends JpaBaseService<Organizations>{
@Autowired
KafkaPersistService kafkaPersistService;
MqPersistService mqPersistService;
public OrganizationsService() {
super(OrganizationsMapper.class);
@@ -51,8 +51,8 @@ public class OrganizationsService extends JpaBaseService<Organizations>{
public boolean insert(Organizations organization) {
if(super.insert(organization)){
kafkaPersistService.send(
KafkaIdentityTopic.ORG_TOPIC, organization, KafkaIdentityAction.CREATE_ACTION);
mqPersistService.send(
MqIdentityTopic.ORG_TOPIC, organization, MqIdentityAction.CREATE_ACTION);
return true;
}
return false;
@@ -60,8 +60,8 @@ public class OrganizationsService extends JpaBaseService<Organizations>{
public boolean update(Organizations organization) {
if(super.update(organization)){
kafkaPersistService.send(
KafkaIdentityTopic.ORG_TOPIC, organization, KafkaIdentityAction.UPDATE_ACTION);
mqPersistService.send(
MqIdentityTopic.ORG_TOPIC, organization, MqIdentityAction.UPDATE_ACTION);
return true;
}
return false;
@@ -82,8 +82,8 @@ public class OrganizationsService extends JpaBaseService<Organizations>{
public boolean delete(Organizations organization) {
if(super.delete(organization)){
kafkaPersistService.send(
KafkaIdentityTopic.ORG_TOPIC, organization, KafkaIdentityAction.DELETE_ACTION);
mqPersistService.send(
MqIdentityTopic.ORG_TOPIC, organization, MqIdentityAction.DELETE_ACTION);
return true;
}
return false;
maxkey-persistence/src/main/java/org/maxkey/persistence/service/UserInfoService.java
+19 -19
View File
@@ -26,10 +26,10 @@ import org.maxkey.crypto.password.PasswordReciprocal;
import org.maxkey.entity.Accounts;
import org.maxkey.entity.ChangePassword;
import org.maxkey.entity.UserInfo;
import org.maxkey.persistence.kafka.KafkaIdentityAction;
import org.maxkey.persistence.kafka.KafkaIdentityTopic;
import org.maxkey.persistence.kafka.KafkaPersistService;
import org.maxkey.persistence.mapper.UserInfoMapper;
import org.maxkey.persistence.mq.MqIdentityAction;
import org.maxkey.persistence.mq.MqIdentityTopic;
import org.maxkey.persistence.mq.MqPersistService;
import org.maxkey.persistence.repository.PasswordPolicyValidator;
import org.maxkey.util.DateUtils;
import org.maxkey.util.StringUtils;
@@ -57,7 +57,7 @@ public class UserInfoService extends JpaBaseService<UserInfo> {
PasswordPolicyValidator passwordPolicyValidator;
@Autowired
KafkaPersistService kafkaPersistService;
MqPersistService mqPersistService;
@Autowired
protected JdbcTemplate jdbcTemplate;
@@ -79,12 +79,12 @@ public class UserInfoService extends JpaBaseService<UserInfo> {
public boolean insert(UserInfo userInfo) {
userInfo = passwordEncoder(userInfo);
if (super.insert(userInfo)) {
if(kafkaPersistService.getApplicationConfig().isKafkaSupport()) {
if(mqPersistService.getApplicationConfig().isMessageQueueSupport()) {
UserInfo loadUserInfo = findUserRelated(userInfo.getId());
kafkaPersistService.send(
KafkaIdentityTopic.USERINFO_TOPIC,
mqPersistService.send(
MqIdentityTopic.USERINFO_TOPIC,
loadUserInfo,
KafkaIdentityAction.CREATE_ACTION);
MqIdentityAction.CREATE_ACTION);
}
return true;
@@ -96,13 +96,13 @@ public class UserInfoService extends JpaBaseService<UserInfo> {
public boolean update(UserInfo userInfo) {
userInfo = passwordEncoder(userInfo);
if (super.update(userInfo)) {
if(kafkaPersistService.getApplicationConfig().isKafkaSupport()) {
if(mqPersistService.getApplicationConfig().isMessageQueueSupport()) {
UserInfo loadUserInfo = findUserRelated(userInfo.getId());
accountUpdate(loadUserInfo);
kafkaPersistService.send(
KafkaIdentityTopic.USERINFO_TOPIC,
mqPersistService.send(
MqIdentityTopic.USERINFO_TOPIC,
loadUserInfo,
KafkaIdentityAction.UPDATE_ACTION);
MqIdentityAction.UPDATE_ACTION);
}
changePasswordProvisioning(userInfo);
@@ -113,15 +113,15 @@ public class UserInfoService extends JpaBaseService<UserInfo> {
public boolean delete(UserInfo userInfo) {
UserInfo loadUserInfo = null;
if(kafkaPersistService.getApplicationConfig().isKafkaSupport()) {
if(mqPersistService.getApplicationConfig().isMessageQueueSupport()) {
loadUserInfo = findUserRelated(userInfo.getId());
}
if( super.delete(userInfo)){
kafkaPersistService.send(
KafkaIdentityTopic.USERINFO_TOPIC,
mqPersistService.send(
MqIdentityTopic.USERINFO_TOPIC,
loadUserInfo,
KafkaIdentityAction.DELETE_ACTION);
MqIdentityAction.DELETE_ACTION);
accountUpdate(loadUserInfo);
return true;
}
@@ -326,10 +326,10 @@ public class UserInfoService extends JpaBaseService<UserInfo> {
changePassword.setDecipherable(loadUserInfo.getDecipherable());
changePassword.setPassword(loadUserInfo.getPassword());
changePassword.setInstId(loadUserInfo.getInstId());
kafkaPersistService.send(
KafkaIdentityTopic.PASSWORD_TOPIC,
mqPersistService.send(
MqIdentityTopic.PASSWORD_TOPIC,
changePassword,
KafkaIdentityAction.PASSWORD_ACTION);
MqIdentityAction.PASSWORD_ACTION);
}
}