Просмотр исходного кода

[消息组件][修改][支持websocket]

release_toaliyun_real
luozukai 7 лет назад
committed by Stormeye Wu
Родитель
Сommit
e6e8df1708
9 измененных файлов: 528 добавлений и 16 удалений
  1. +365
    -0
      mallinkAdmin/src/main/java/BinarySearchTree.java
  2. +0
    -1
      mallinkMQConsumer/src/main/java/com/iformall/mq/MqBaseConsumer.java
  3. +12
    -0
      mallinkService/src/main/java/com/iformall/domain/po/BaseMsg.java
  4. +0
    -10
      mallinkService/src/main/java/com/iformall/domain/po/WxMsgRecord.java
  5. +40
    -0
      mallinkService/src/main/java/com/iformall/enums/EnumMsgRecordDomain.java
  6. +2
    -2
      mallinkWebSocketServer/src/main/java/com/iformall/controller/SendController.java
  7. +64
    -0
      mallinkWebSocketServer/src/main/java/com/iformall/mq/AmqpMsgConsumer.java
  8. +42
    -0
      mallinkWebSocketServer/src/main/java/com/iformall/mq/impl/RabbitMqAmqpMsgConsumer.java
  9. +3
    -3
      mallinkWebSocketServer/src/main/resources/application-dev.yml

+ 365
- 0
mallinkAdmin/src/main/java/BinarySearchTree.java Просмотреть файл

@@ -0,0 +1,365 @@
import java.util.ArrayList;
import java.util.List;

public class BinarySearchTree {

// 树的根结点
private TreeNode root = null;

// 遍历结点列表
private List<TreeNode> nodelist = new ArrayList<>();

// 当前结点
private class TreeNode {
private int key;
private TreeNode leftChild;
private TreeNode rightChild;
private TreeNode parent;

public TreeNode(int key, TreeNode leftChild, TreeNode rightChild,
TreeNode parent) {
this.key = key;
this.leftChild = leftChild;
this.rightChild = rightChild;
this.parent = parent;
}

public int getKey() {
return key;
}

public String toString() {
String leftkey = (leftChild == null ? "" : String
.valueOf(leftChild.key));
String rightkey = (rightChild == null ? "" : String
.valueOf(rightChild.key));
return "(" + leftkey + " , " + key + " , " + rightkey + ")";
}

}

/**
* isEmpty: 判断二叉查找树是否为空;若为空,返回 true ,否则返回 false .
*
*/
public boolean isEmpty() {
if (root == null) {
return true;
} else {
return false;
}
}

/**
* TreeEmpty: 对于某些二叉查找树操作(比如删除关键字)来说,若树为空,则抛出异常。
*/
public void TreeEmpty() throws Exception {
if (isEmpty()) {
throw new Exception("树为空!");
}
}

/**
* search: 在二叉查找树中查询给定关键字
*
* @param key
* 给定关键字
* @return 匹配给定关键字的树结点
*/
public TreeNode search(int key) {
TreeNode pNode = root;
while (pNode != null && pNode.key != key) {
if (key < pNode.key) {
pNode = pNode.leftChild;
} else {
pNode = pNode.rightChild;
}
}
return pNode;
}

/**
* minElemNode: 获取二叉查找树中的最小关键字结点
*
* @return 二叉查找树的最小关键字结点
* @throws Exception
* 若树为空,则抛出异常
*/
public TreeNode minElemNode(TreeNode node) throws Exception {
if (node == null) {
throw new Exception("树为空!");
}
TreeNode pNode = node;
while (pNode.leftChild != null) {
pNode = pNode.leftChild;
}
return pNode;
}

/**
* maxElemNode: 获取二叉查找树中的最大关键字结点
*
* @return 二叉查找树的最大关键字结点
* @throws Exception
* 若树为空,则抛出异常
*/
public TreeNode maxElemNode(TreeNode node) throws Exception {
if (node == null) {
throw new Exception("树为空!");
}
TreeNode pNode = node;
while (pNode.rightChild != null) {
pNode = pNode.rightChild;
}
return pNode;
}

/**
* successor: 获取给定结点在中序遍历顺序下的后继结点
*
* @param node
* 给定树中的结点
* @return 若该结点存在中序遍历顺序下的后继结点,则返回其后继结点;否则返回 null
* @throws Exception
*/
public TreeNode successor(TreeNode node) throws Exception {
if (node == null) {
return null;
}

// 若该结点的右子树不为空,则其后继结点就是右子树中的最小关键字结点
if (node.rightChild != null) {
return minElemNode(node.rightChild);
}
// 若该结点右子树为空
TreeNode parentNode = node.parent;
while (parentNode != null && node == parentNode.rightChild) {
node = parentNode;
parentNode = parentNode.parent;
}
return parentNode;
}

/**
* precessor: 获取给定结点在中序遍历顺序下的前趋结点
*
* @param node
* 给定树中的结点
* @return 若该结点存在中序遍历顺序下的前趋结点,则返回其前趋结点;否则返回 null
* @throws Exception
*/
public TreeNode precessor(TreeNode node) throws Exception {
if (node == null) {
return null;
}

// 若该结点的左子树不为空,则其前趋结点就是左子树中的最大关键字结点
if (node.leftChild != null) {
return maxElemNode(node.leftChild);
}
// 若该结点左子树为空
TreeNode parentNode = node.parent;
while (parentNode != null && node == parentNode.leftChild) {
node = parentNode;
parentNode = parentNode.parent;
}
return parentNode;
}

/**
* insert: 将给定关键字插入到二叉查找树中
*
* @param key
* 给定关键字
*/
public void insert(int key) {
TreeNode parentNode = null;
TreeNode newNode = new TreeNode(key, null, null, null);
TreeNode pNode = root;
if (root == null) {
root = newNode;
return;
}
while (pNode != null) {
parentNode = pNode;
if (key < pNode.key) {
pNode = pNode.leftChild;
} else if (key > pNode.key) {
pNode = pNode.rightChild;
} else {
// 树中已存在匹配给定关键字的结点,则什么都不做直接返回
return;
}
}
if (key < parentNode.key) {
parentNode.leftChild = newNode;
newNode.parent = parentNode;
} else {
parentNode.rightChild = newNode;
newNode.parent = parentNode;
}

}

/**
* insert: 从二叉查找树中删除匹配给定关键字相应的树结点
*
* @param key
* 给定关键字
*/
public void delete(int key) throws Exception {
TreeNode pNode = search(key);
if (pNode == null) {
throw new Exception("树中不存在要删除的关键字!");
}
delete(pNode);
}

/**
* delete: 从二叉查找树中删除给定的结点.
*
* @param pNode
* 要删除的结点
*
* 前置条件: 给定结点在二叉查找树中已经存在
* @throws Exception
*/
private void delete(TreeNode pNode) throws Exception {
if (pNode == null) {
return;
}
if (pNode.leftChild == null && pNode.rightChild == null) { // 该结点既无左孩子结点,也无右孩子结点
TreeNode parentNode = pNode.parent;
if (pNode == parentNode.leftChild) {
parentNode.leftChild = null;
} else {
parentNode.rightChild = null;
}
return;
}
if (pNode.leftChild == null && pNode.rightChild != null) { // 该结点左孩子结点为空,右孩子结点非空
TreeNode parentNode = pNode.parent;
if (pNode == parentNode.leftChild) {
parentNode.leftChild = pNode.rightChild;
pNode.rightChild.parent = parentNode;
} else {
parentNode.rightChild = pNode.rightChild;
pNode.rightChild.parent = parentNode;
}
return;
}
if (pNode.leftChild != null && pNode.rightChild == null) { // 该结点左孩子结点非空,右孩子结点为空
TreeNode parentNode = pNode.parent;
if (pNode == parentNode.leftChild) {
parentNode.leftChild = pNode.leftChild;
pNode.rightChild.parent = parentNode;
} else {
parentNode.rightChild = pNode.leftChild;
pNode.rightChild.parent = parentNode;
}
return;
}
// 该结点左右孩子结点均非空,则删除该结点的后继结点,并用该后继结点取代该结点
TreeNode successorNode = successor(pNode);
delete(successorNode);
pNode.key = successorNode.key;
}

/**
* inOrderTraverseList: 获得二叉查找树的中序遍历结点列表
*
* @return 二叉查找树的中序遍历结点列表
*/
public List<TreeNode> inOrderTraverseList() {
if (nodelist != null) {
nodelist.clear();
}
inOrderTraverse(root);
return nodelist;
}

/**
* inOrderTraverse: 对给定二叉查找树进行中序遍历
*
* @param root
* 给定二叉查找树的根结点
*/
private void inOrderTraverse(TreeNode root) {
if (root != null) {
inOrderTraverse(root.leftChild);
nodelist.add(root);
inOrderTraverse(root.rightChild);
}
}

/**
* toStringOfOrderList: 获取二叉查找树中关键字的有序列表
*
* @return 二叉查找树中关键字的有序列表
*/
public String toStringOfOrderList() {
StringBuilder sbBuilder = new StringBuilder(" [ ");
for (TreeNode p : inOrderTraverseList()) {
sbBuilder.append(p.key);
sbBuilder.append(" ");
}
sbBuilder.append("]");
return sbBuilder.toString();
}

/**
* 获取该二叉查找树的字符串表示
*/
public String toString() {
StringBuilder sbBuilder = new StringBuilder(" [ ");
for (TreeNode p : inOrderTraverseList()) {
sbBuilder.append(p);
sbBuilder.append(" ");
}
sbBuilder.append("]");
return sbBuilder.toString();
}

public TreeNode getRoot() {
return root;
}

public static void testNode(BinarySearchTree bst, TreeNode pNode)
throws Exception {
System.out.println("本结点: " + pNode);
System.out.println("前趋结点: " + bst.precessor(pNode));
System.out.println("后继结点: " + bst.successor(pNode));
}

public static void testTraverse(BinarySearchTree bst) {
System.out.println("二叉树遍历:" + bst);
System.out.println("二叉查找树转换为有序列表: " + bst.toStringOfOrderList());
}

public static void main(String[] args) {
try {
BinarySearchTree bst = new BinarySearchTree();
System.out.println("查找树是否为空? " + (bst.isEmpty() ? "是" : "否"));
int[] keys = new int[] { 15, 6, 18, 3, 7, 13, 20, 2, 9, 4 };
for (int key : keys) {
bst.insert(key);
}
System.out.println("查找树是否为空? " + (bst.isEmpty() ? "是" : "否"));
TreeNode minkeyNode = bst.minElemNode(bst.getRoot());
System.out.println("最小关键字: " + minkeyNode.getKey());
testNode(bst, minkeyNode);
TreeNode maxKeyNode = bst.maxElemNode(bst.getRoot());
System.out.println("最大关键字: " + maxKeyNode.getKey());
testNode(bst, maxKeyNode);
System.out.println("根结点关键字: " + bst.getRoot().getKey());
testNode(bst, bst.getRoot());
testTraverse(bst);
System.out.println("****************************** ");
testTraverse(bst);
} catch (Exception e) {
System.out.println(e.getMessage());
e.printStackTrace();
}
}

}

+ 0
- 1
mallinkMQConsumer/src/main/java/com/iformall/mq/MqBaseConsumer.java Просмотреть файл

@@ -52,7 +52,6 @@ public class MqBaseConsumer {
//微信小程序
sendSmartAppMsgService.send((SmartAppMsg)JsonUtil.readValue(message,SmartAppMsg.class));
}

baseMsg.setMsgStatus(EnumMsgRecordStatus.CONSUME_SUCC.getCode());
wxMsgRecordMapper.update(baseMsg);
}catch (Exception e){


+ 12
- 0
mallinkService/src/main/java/com/iformall/domain/po/BaseMsg.java Просмотреть файл

@@ -9,6 +9,10 @@ import java.io.Serializable;
*/
public class BaseMsg implements Serializable {
private static final long serialVersionUID = 923764283764823L;

//域0全部 1sys 2a端 3c端 4b端
private Integer domain = 0;

//租户id
private String tenantId;

@@ -37,6 +41,14 @@ public class BaseMsg implements Serializable {
private Integer delayTimeLevel = 0;


public Integer getDomain() {
return domain;
}

public void setDomain(Integer domain) {
this.domain = domain;
}

public String getTenantId() {
return tenantId;
}


+ 0
- 10
mallinkService/src/main/java/com/iformall/domain/po/WxMsgRecord.java Просмотреть файл

@@ -13,8 +13,6 @@ public class WxMsgRecord extends BaseMsg {
protected Long id;
//租户id
private String tenantId;
//域1sys 2a端 3c端 4b端
private Integer domain;
//msg 消息类型1短信 2回调短信 邮件 3微信小程序模板 4微信公众号模板 5系统通
private String msg;
//发送时间
@@ -93,14 +91,6 @@ public class WxMsgRecord extends BaseMsg {
this.id = id;
}

public Integer getDomain() {
return domain;
}

public void setDomain(Integer domain) {
this.domain = domain;
}

public String getSender() {
return sender;
}


+ 40
- 0
mallinkService/src/main/java/com/iformall/enums/EnumMsgRecordDomain.java Просмотреть файл

@@ -0,0 +1,40 @@
package com.iformall.enums;

/**
* Created by luozukai
* 域:0全部1sys 2a端 3c端 4b端
*/
public enum EnumMsgRecordDomain {
ALL(0, "全部"),
SYS(2, "系统消息"),
A(3, "a端"),
C(4, "c端"),
B(5, "b端"),
USER(6, "用户"),
;

public static EnumMsgRecordDomain getEnum(Integer code) {
for (EnumMsgRecordDomain value : values()) {
if (value.getCode().equals(code)) {
return value;
}
}
return null;
}

private Integer code;
private String message;

EnumMsgRecordDomain(Integer code, String message) {
this.code = code;
this.message = message;
}

public Integer getCode() {
return code;
}

public String getMessage() {
return message;
}
}

+ 2
- 2
mallinkWebSocketServer/src/main/java/com/iformall/controller/SendController.java Просмотреть файл

@@ -59,8 +59,8 @@ public class SendController {
// 方式1
messagingTemplate.convertAndSendToUser(name, "/message", JSON.toJSONString(message));
// 方式2
//String userId = userService.getSessionId(name);
//amqpTemplate.convertAndSend("", Constance.queue_pre+userId, message);
String userId = userService.getSessionId(name);
amqpTemplate.convertAndSend("", Constance.queue_pre+userId, message);
}




+ 64
- 0
mallinkWebSocketServer/src/main/java/com/iformall/mq/AmqpMsgConsumer.java Просмотреть файл

@@ -0,0 +1,64 @@
package com.iformall.mq;

import com.alibaba.fastjson.JSON;
import com.iformall.domain.po.*;
import com.iformall.enums.EnumMsgRecordDomain;
import com.iformall.enums.EnumMsgRecordStatus;
import com.iformall.mapper.WxMsgRecordMapper;
import com.iformall.service.UserService;
import com.iformall.util.Constance;
import com.iformall.utils.JsonUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.simp.SimpMessagingTemplate;
import org.springframework.stereotype.Service;

/**
* 消息消费入口
*/
@Service
public class AmqpMsgConsumer {
private final Logger log = LoggerFactory.getLogger(this.getClass());
@Autowired
private WxMsgRecordMapper wxMsgRecordMapper;
@Autowired
private AmqpTemplate amqpTemplate;
@Autowired
private SimpMessagingTemplate messagingTemplate;
@Autowired
private UserService userService;

public void doMessage(String message) {
log.info("received amqp message: {}", message);
BaseMsg baseMsg = null;

try {
baseMsg = (BaseMsg)JsonUtil.readValue(message,BaseMsg.class);

if(EnumMsgRecordDomain.A.getCode().equals(baseMsg.getDomain())) {
messagingTemplate.convertAndSend("/topic/" + baseMsg.getTenantId() + "A", message);
}else if(EnumMsgRecordDomain.B.getCode().equals(baseMsg.getDomain())) {
messagingTemplate.convertAndSend("/topic/" + baseMsg.getTenantId() + "B", message);
}else if(EnumMsgRecordDomain.C.getCode().equals(baseMsg.getDomain())) {
messagingTemplate.convertAndSend("/topic/" + baseMsg.getTenantId() + "C",message);
}else if(EnumMsgRecordDomain.USER.getCode().equals(baseMsg.getDomain())) {
// 方式1
messagingTemplate.convertAndSendToUser(baseMsg.getTenantId()+baseMsg.getReceiver(), "/message", message);
// 方式2
// String userId = userService.getSessionId(name);
// amqpTemplate.convertAndSend("", Constance.queue_pre+userId, message);
}

baseMsg.setMsgStatus(EnumMsgRecordStatus.CONSUME_SUCC.getCode());
wxMsgRecordMapper.update(baseMsg);
}catch (Exception e){
log.error("consum received error: {}", e);
baseMsg.setStatusMessage(e.getMessage());
baseMsg.setMsgStatus(EnumMsgRecordStatus.CONSUME_FAIL.getCode());
wxMsgRecordMapper.update(baseMsg);
}

}
}

+ 42
- 0
mallinkWebSocketServer/src/main/java/com/iformall/mq/impl/RabbitMqAmqpMsgConsumer.java Просмотреть файл

@@ -0,0 +1,42 @@
package com.iformall.mq.impl;

import com.iformall.mq.AmqpMsgConsumer;
import com.iformall.mq.MQConfig;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Profile;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

@Component
@Profile(MQConfig.Impl.RABBIT_MQ)
public class RabbitMqAmqpMsgConsumer extends AmqpMsgConsumer {
String queueName = "topic-amqp";

@Autowired
private AmqpAdmin amqpAdmin;

@PostConstruct
public void init() {
DirectExchange exchange = new DirectExchange(queueName);
Queue queue = new Queue(queueName);
Binding binding = BindingBuilder.bind(queue).to(exchange).withQueueName();
amqpAdmin.declareExchange(exchange);
amqpAdmin.declareQueue(queue);
amqpAdmin.declareBinding(binding);
}

@Autowired
private AmqpTemplate rabbitTemplate;

public void send(String msg) {
rabbitTemplate.convertAndSend(queueName, msg);
}

@RabbitListener(queues = "topic-amqp")
public void onMessage(String message) {
doMessage(message);
}
}

+ 3
- 3
mallinkWebSocketServer/src/main/resources/application-dev.yml Просмотреть файл

@@ -61,10 +61,10 @@ spring:
retry-another-broker-when-not-store-ok: false
retry-times-when-send-failed: 2
rabbitmq:
host: localhost
host: 202.165.179.86
port: 5672
username: guest
password: guest
username: fumao
password: f8l89&*%%u7f1t22
publisher-confirms: true
virtual-host: /



Загрузка…
Отмена
Сохранить