[大数据]JAVA语言(Mysql触发器)

Mysql触发器监听A库中某一张表的动态变化，通过调用jar包的方式，调用另一个项目服务，使B库中的数据改变；或是直接将数据返回给前端。
注：（1）多数据源配置
	（2）java在后台建立一个websocket，页面与此socket建立连接，
    （3）写一个jar程序，当运行时与socket建立连接，并将参数发送到此socket
    （4）数据库建立insert、update、delete触发器，当数据改变时调用外部jar程序

多数据源配置：https://blog.csdn.net/w57685321/article/details/106823660/

触发器是与表有关的数据库对象，在满足定义条件时触发，并执行触发器中定义的语句集合。触发器的这种特性可以协助应用在数据库端确保数据的完整性。

mysql触发器trigger 实例详解 ：https://www.cnblogs.com/phpper/p/7587031.html

BEFORE和AFTER参数指定了触发执行的时间，在事件之前或是之后
FOR EACH ROW表示任何一条记录上的操作满足触发事件都会触发该触发器

#创建有多个执行语句的触发器
CREATE TRIGGER 触发器名 BEFORE|AFTER 触发事件
ON 表名 FOR EACH ROW
BEGIN
    执行语句列表
END

1.创建表demo表结构如下图

2.创建触发器

drop TRIGGER insertTrigger; 				-- 删除触发器

CREATE TRIGGER insertTrigger after insert 	-- insertTrigger 触发器的名称
 ON demo FOR EACH row    					-- 该触发器是在demo表改变的时候发生				
begin
DECLARE redata INT;    						 -- 定义一个变量接收Select sys_exec...语句执行完成后的结果
    Select sys_exec('java -jar D:\\logs\\sql-trigger-1.0.jar "http://localhost:8080/project/test" "demo" "insert"') INTO redata; 
end

-- 要调用的外部服务打成jar:sql-trigger-1.0.jar
-- jar包存放的地址：D:\\logs\\jar包全名称
-- 需要调用另一个项目服务中的方法：http://localhost:8080/project/test
-- 传递的参数：demo、insert

jar包程序，创建maven项目，

package com.cctegitc;
import com.cctegitc.util.HttpUtil;
/**
 * @Description:启动main方法
 * @Author: yrc
 * @Date: 2022/3/29 13:52
 * @Param: [args]
 * @Return: void 
 * @Throws:    
 */
public class Main {
    public static void main(String[] args) {
        if (args != null && args.length >= 3) {
            HttpUtil.request(args[0], args[1], args[2]);
        }
    }
}
//---------------------------------------------------------------------------------------------------
package com.cctegitc.util;
import java.io.IOException;

import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
/**
 * @Description:触发器调用后台请求
 * @Author: yrc
 * @Date: 2022/3/29 13:53
 * @Param: 
 * @Return:  
 * @Throws:    
 */
public class HttpUtil {
    public static void request(String url, String table, String action) {
        // httpclient客户端，类似于一个浏览器，可以由这个客户端执行http请求
        CloseableHttpClient httpClient = HttpClients.createDefault();
        // 请求
        HttpGet httpGet = new HttpGet(url + "?table="+table + "&action=" + action);
        // 响应
        try {
            // execute()执行成功会返回HttpResponse响应
            httpClient.execute(httpGet);
            // 响应体
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            try {
                httpClient.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
//配置文件：config.properties
url=http://localhost:8080/sqlTrigger
//pom.xml文件配置
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.cctegitc</groupId>
    <artifactId>sql-trigger</artifactId>
    <version>1.0</version>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.httpcomponents</groupId>
            <artifactId>httpclient</artifactId>
            <version>4.5.13</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <version>2.4</version>
                <configuration>
                    <appendAssemblyId>false</appendAssemblyId>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                    <archive>
                        <manifest>
                            <mainClass>com.cctegitc.Main</mainClass>
                        </manifest>
                    </archive>
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

</project>

mysql中没有调用外部脚本的功能：
需要安装lib_mysqludf_sys调用外部脚本： https://blog.csdn.net/miaodichiyou/article/details/102566958

3.java后台代码：创建一个websocket

import java.io.IOException;
import java.util.concurrent.CopyOnWriteArraySet;
 
import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
 
/**
 * @ServerEndpoint 注解是一个类层次的注解，它的功能主要是将目前的类定义成一个websocket服务器端,
 * 注解的值将被用于监听用户连接的终端访问URL地址,客户端可以通过这个URL来连接到WebSocket服务器端
 */
@ServerEndpoint("/WSwebsocket")
public class WebSocketTest {
	//静态变量，用来记录当前在线连接数。应该把它设计成线程安全的。
	private static int onlineCount = 0;
 
	//concurrent包的线程安全Set，用来存放每个客户端对应的MyWebSocket对象。若要实现服务端与单一客户端通信的话，可以使用Map来存放，其中Key可以为用户标识
	private static CopyOnWriteArraySet<WebSocketTest> webSocketSet = new CopyOnWriteArraySet<WebSocketTest>();
 
	//与某个客户端的连接会话，需要通过它来给客户端发送数据
	private Session session;
 
	/**
	 * 连接建立成功调用的方法
	 * @param session  可选的参数。session为与某个客户端的连接会话，需要通过它来给客户端发送数据
	 */
	@OnOpen
	public void onOpen(Session session){
		this.session = session;
		webSocketSet.add(this);     //加入set中
		addOnlineCount();           //在线数加1
		System.out.println("有新连接加入！当前在线人数为" + getOnlineCount());
	}
 
	/**
	 * 连接关闭调用的方法
	 */
	@OnClose
	public void onClose(){
		webSocketSet.remove(this);  //从set中删除
		subOnlineCount();           //在线数减1
		System.out.println("有一连接关闭！当前在线人数为" + getOnlineCount());
	}
 
	/**
	 * 收到客户端消息后调用的方法
	 * @param message 客户端发送过来的消息
	 * @param session 可选的参数
	 */
	@OnMessage
	public void onMessage(String message, Session session) {
		System.out.println("来自客户端的消息:" + message);
		//群发消息
		for(WebSocketTest item: webSocketSet){
			try {
				item.sendMessage(message);
			} catch (IOException e) {
				e.printStackTrace();
				continue;
			}
		}
	}
 
	/**
	 * 发生错误时调用
	 * @param session
	 * @param error
	 */
	@OnError
	public void onError(Session session, Throwable error){
		System.out.println("发生错误");
		error.printStackTrace();
	}
 
	/**
	 * 这个方法与上面几个方法不一样。没有用注解，是根据自己需要添加的方法。
	 * @param message
	 * @throws IOException
	 */
	public void sendMessage(String message) throws IOException{
		this.session.getBasicRemote().sendText(message);
		//this.session.getAsyncRemote().sendText(message);
	}
 
	public static synchronized int getOnlineCount() {
		return onlineCount;
	}
 
	public static synchronized void addOnlineCount() {
		WebSocketTest.onlineCount++;
	}
 
	public static synchronized void subOnlineCount() {
		WebSocketTest.onlineCount--;
	}
}

4.页面代码：链接此websocket

<%@ page language="java" pageEncoding="UTF-8" %>
<!DOCTYPE html>
<html>
<head>
    <title>Java后端WebSocket的Tomcat实现</title>
</head>
<body>
    Welcome<br/><input id="text" type="text"/>
    <button οnclick="send()">发送消息</button>
    <hr/>
    <button οnclick="closeWebSocket()">关闭WebSocket连接</button>
    <hr/>
    <div id="message"></div>
</body>
 
<script type="text/javascript">
    var websocket = null;
    //判断当前浏览器是否支持WebSocket
    if ('WebSocket' in window) {
        websocket = new WebSocket("ws://192.168.1.29:8082/bim/WSwebsocket");
    }
    else {
        alert('当前浏览器 Not support websocket')
    }
 
    //连接发生错误的回调方法
    websocket.onerror = function () {
        setMessageInnerHTML("WebSocket连接发生错误");
    };
 
    //连接成功建立的回调方法
    websocket.onopen = function () {
        setMessageInnerHTML("WebSocket连接成功");
    }
 
    //接收到消息的回调方法
    websocket.onmessage = function (event) {
        setMessageInnerHTML(event.data);
    }
 
    //连接关闭的回调方法
    websocket.onclose = function () {
        setMessageInnerHTML("WebSocket连接关闭");
    }
 
    //监听窗口关闭事件，当窗口关闭时，主动去关闭websocket连接，防止连接还没断开就关闭窗口，server端会抛异常。
    window.onbeforeunload = function () {
        closeWebSocket();
    }
 
    //将消息显示在网页上
    function setMessageInnerHTML(innerHTML) {
        document.getElementById('message').innerHTML += innerHTML + '<br/>';
    }
 
    //关闭WebSocket连接
    function closeWebSocket() {
        websocket.close();
    }
 
    //发送消息
    function send() {
        var message = document.getElementById('text').value;
        websocket.send(message);
    }
</script>
</html>