瀏覽代碼

zhaoshang project data enrypt and decode data init

wwd 3 年之前
父節點
當前提交
41106881fb
共有 34 個文件被更改,包括 2209 次插入10 次删除
  1. 29 10
      .gitignore
  2. 118 0
      .mvn/wrapper/MavenWrapperDownloader.java
  3. 二進制
      .mvn/wrapper/maven-wrapper.jar
  4. 2 0
      .mvn/wrapper/maven-wrapper.properties
  5. 310 0
      mvnw
  6. 182 0
      mvnw.cmd
  7. 192 0
      pom.xml
  8. 26 0
      src/main/java/com/persagy/ztkencryptdecodedata/SystemConfig.java
  9. 23 0
      src/main/java/com/persagy/ztkencryptdecodedata/ZtkEncryptdecodedataApplication.java
  10. 27 0
      src/main/java/com/persagy/ztkencryptdecodedata/config/Base64Util.java
  11. 100 0
      src/main/java/com/persagy/ztkencryptdecodedata/config/RSAUtil.java
  12. 34 0
      src/main/java/com/persagy/ztkencryptdecodedata/config/YmlPropertySourceFactory.java
  13. 35 0
      src/main/java/com/persagy/ztkencryptdecodedata/controller/TestEnDeServiceController.java
  14. 21 0
      src/main/java/com/persagy/ztkencryptdecodedata/dataSafety/DataOptRSA.java
  15. 6 0
      src/main/java/com/persagy/ztkencryptdecodedata/dataSafety/DecrypInputMessageInterface.java
  16. 62 0
      src/main/java/com/persagy/ztkencryptdecodedata/dataSafety/DecryptInputMessageService.java
  17. 6 0
      src/main/java/com/persagy/ztkencryptdecodedata/dataSafety/EncryptInputMessageInterface.java
  18. 74 0
      src/main/java/com/persagy/ztkencryptdecodedata/dataSafety/EncryptInputMessageService.java
  19. 31 0
      src/main/java/com/persagy/ztkencryptdecodedata/entity/ChattingUser.java
  20. 20 0
      src/main/java/com/persagy/ztkencryptdecodedata/entity/SecretKeyConfig.java
  21. 139 0
      src/main/java/com/persagy/ztkencryptdecodedata/kafka/CloudKafkaConsumerFromEdgeTopic.java
  22. 132 0
      src/main/java/com/persagy/ztkencryptdecodedata/kafka/EdgeKafkaProducer.java
  23. 7 0
      src/main/java/com/persagy/ztkencryptdecodedata/netty/NettyClientInterface.java
  24. 172 0
      src/main/java/com/persagy/ztkencryptdecodedata/netty/NettyClinet.java
  25. 8 0
      src/main/java/com/persagy/ztkencryptdecodedata/netty/NettyConstants.java
  26. 235 0
      src/main/java/com/persagy/ztkencryptdecodedata/netty/NettyServer.java
  27. 19 0
      src/main/java/com/persagy/ztkencryptdecodedata/netty/TcpClientHandler.java
  28. 33 0
      src/main/resources/application-dev.yml
  29. 33 0
      src/main/resources/application-prod.yml
  30. 33 0
      src/main/resources/application-uat.yml
  31. 7 0
      src/main/resources/application.yml
  32. 57 0
      src/test/java/com/persagy/ztkencryptdecodedata/TcpServer.java
  33. 23 0
      src/test/java/com/persagy/ztkencryptdecodedata/TcpServerHandler.java
  34. 13 0
      src/test/java/com/persagy/ztkencryptdecodedata/ZtkEncryptdecodedataApplicationTests.java

+ 29 - 10
.gitignore

@@ -1,14 +1,33 @@
-# ---> Java
-*.class
+HELP.md
+target/
+!.mvn/wrapper/maven-wrapper.jar
+!**/src/main/**/target/
+!**/src/test/**/target/
 
-# Mobile Tools for Java (J2ME)
-.mtj.tmp/
+### STS ###
+.apt_generated
+.classpath
+.factorypath
+.project
+.settings
+.springBeans
+.sts4-cache
 
-# Package Files #
-*.jar
-*.war
-*.ear
+### IntelliJ IDEA ###
+.idea
+*.iws
+*.iml
+*.ipr
 
-# virtual machine crash logs, see http://www.java.com/en/download/help/error_hotspot.xml
-hs_err_pid*
+### NetBeans ###
+/nbproject/private/
+/nbbuild/
+/dist/
+/nbdist/
+/.nb-gradle/
+build/
+!**/src/main/**/build/
+!**/src/test/**/build/
 
+### VS Code ###
+.vscode/

+ 118 - 0
.mvn/wrapper/MavenWrapperDownloader.java

@@ -0,0 +1,118 @@
+/*
+ * Copyright 2007-present the original author or authors.
+ *
+ * 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
+ *
+ *      https://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.
+ */
+
+import java.net.*;
+import java.io.*;
+import java.nio.channels.*;
+import java.util.Properties;
+
+public class MavenWrapperDownloader {
+
+    private static final String WRAPPER_VERSION = "0.5.6";
+    /**
+     * Default URL to download the maven-wrapper.jar from, if no 'downloadUrl' is provided.
+     */
+    private static final String DEFAULT_DOWNLOAD_URL = "https://repo.maven.apache.org/maven2/io/takari/maven-wrapper/"
+            + WRAPPER_VERSION + "/maven-wrapper-" + WRAPPER_VERSION + ".jar";
+
+    /**
+     * Path to the maven-wrapper.properties file, which might contain a downloadUrl property to
+     * use instead of the default one.
+     */
+    private static final String MAVEN_WRAPPER_PROPERTIES_PATH =
+            ".mvn/wrapper/maven-wrapper.properties";
+
+    /**
+     * Path where the maven-wrapper.jar will be saved to.
+     */
+    private static final String MAVEN_WRAPPER_JAR_PATH =
+            ".mvn/wrapper/maven-wrapper.jar";
+
+    /**
+     * Name of the property which should be used to override the default download url for the wrapper.
+     */
+    private static final String PROPERTY_NAME_WRAPPER_URL = "wrapperUrl";
+
+    public static void main(String args[]) {
+        System.out.println("- Downloader started");
+        File baseDirectory = new File(args[0]);
+        System.out.println("- Using base directory: " + baseDirectory.getAbsolutePath());
+
+        // If the maven-wrapper.properties exists, read it and check if it contains a custom
+        // wrapperUrl parameter.
+        File mavenWrapperPropertyFile = new File(baseDirectory, MAVEN_WRAPPER_PROPERTIES_PATH);
+        String url = DEFAULT_DOWNLOAD_URL;
+        if (mavenWrapperPropertyFile.exists()) {
+            FileInputStream mavenWrapperPropertyFileInputStream = null;
+            try {
+                mavenWrapperPropertyFileInputStream = new FileInputStream(mavenWrapperPropertyFile);
+                Properties mavenWrapperProperties = new Properties();
+                mavenWrapperProperties.load(mavenWrapperPropertyFileInputStream);
+                url = mavenWrapperProperties.getProperty(PROPERTY_NAME_WRAPPER_URL, url);
+            } catch (IOException e) {
+                System.out.println("- ERROR loading '" + MAVEN_WRAPPER_PROPERTIES_PATH + "'");
+            } finally {
+                try {
+                    if (mavenWrapperPropertyFileInputStream != null) {
+                        mavenWrapperPropertyFileInputStream.close();
+                    }
+                } catch (IOException e) {
+                    // Ignore ...
+                }
+            }
+        }
+        System.out.println("- Downloading from: " + url);
+
+        File outputFile = new File(baseDirectory.getAbsolutePath(), MAVEN_WRAPPER_JAR_PATH);
+        if (!outputFile.getParentFile().exists()) {
+            if (!outputFile.getParentFile().mkdirs()) {
+                System.out.println(
+                        "- ERROR creating output directory '" + outputFile.getParentFile().getAbsolutePath() + "'");
+            }
+        }
+        System.out.println("- Downloading to: " + outputFile.getAbsolutePath());
+        try {
+            downloadFileFromURL(url, outputFile);
+            System.out.println("Done");
+            System.exit(0);
+        } catch (Throwable e) {
+            System.out.println("- Error downloading");
+            e.printStackTrace();
+            System.exit(1);
+        }
+    }
+
+    private static void downloadFileFromURL(String urlString, File destination) throws Exception {
+        if (System.getenv("MVNW_USERNAME") != null && System.getenv("MVNW_PASSWORD") != null) {
+            String username = System.getenv("MVNW_USERNAME");
+            char[] password = System.getenv("MVNW_PASSWORD").toCharArray();
+            Authenticator.setDefault(new Authenticator() {
+                @Override
+                protected PasswordAuthentication getPasswordAuthentication() {
+                    return new PasswordAuthentication(username, password);
+                }
+            });
+        }
+        URL website = new URL(urlString);
+        ReadableByteChannel rbc;
+        rbc = Channels.newChannel(website.openStream());
+        FileOutputStream fos = new FileOutputStream(destination);
+        fos.getChannel().transferFrom(rbc, 0, Long.MAX_VALUE);
+        fos.close();
+        rbc.close();
+    }
+
+}

二進制
.mvn/wrapper/maven-wrapper.jar


+ 2 - 0
.mvn/wrapper/maven-wrapper.properties

@@ -0,0 +1,2 @@
+distributionUrl=https://repo.maven.apache.org/maven2/org/apache/maven/apache-maven/3.8.1/apache-maven-3.8.1-bin.zip
+wrapperUrl=https://repo.maven.apache.org/maven2/io/takari/maven-wrapper/0.5.6/maven-wrapper-0.5.6.jar

+ 310 - 0
mvnw

@@ -0,0 +1,310 @@
+#!/bin/sh
+# ----------------------------------------------------------------------------
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you 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
+#
+#    https://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.
+# ----------------------------------------------------------------------------
+
+# ----------------------------------------------------------------------------
+# Maven Start Up Batch script
+#
+# Required ENV vars:
+# ------------------
+#   JAVA_HOME - location of a JDK home dir
+#
+# Optional ENV vars
+# -----------------
+#   M2_HOME - location of maven2's installed home dir
+#   MAVEN_OPTS - parameters passed to the Java VM when running Maven
+#     e.g. to debug Maven itself, use
+#       set MAVEN_OPTS=-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=8000
+#   MAVEN_SKIP_RC - flag to disable loading of mavenrc files
+# ----------------------------------------------------------------------------
+
+if [ -z "$MAVEN_SKIP_RC" ] ; then
+
+  if [ -f /etc/mavenrc ] ; then
+    . /etc/mavenrc
+  fi
+
+  if [ -f "$HOME/.mavenrc" ] ; then
+    . "$HOME/.mavenrc"
+  fi
+
+fi
+
+# OS specific support.  $var _must_ be set to either true or false.
+cygwin=false;
+darwin=false;
+mingw=false
+case "`uname`" in
+  CYGWIN*) cygwin=true ;;
+  MINGW*) mingw=true;;
+  Darwin*) darwin=true
+    # Use /usr/libexec/java_home if available, otherwise fall back to /Library/Java/Home
+    # See https://developer.apple.com/library/mac/qa/qa1170/_index.html
+    if [ -z "$JAVA_HOME" ]; then
+      if [ -x "/usr/libexec/java_home" ]; then
+        export JAVA_HOME="`/usr/libexec/java_home`"
+      else
+        export JAVA_HOME="/Library/Java/Home"
+      fi
+    fi
+    ;;
+esac
+
+if [ -z "$JAVA_HOME" ] ; then
+  if [ -r /etc/gentoo-release ] ; then
+    JAVA_HOME=`java-config --jre-home`
+  fi
+fi
+
+if [ -z "$M2_HOME" ] ; then
+  ## resolve links - $0 may be a link to maven's home
+  PRG="$0"
+
+  # need this for relative symlinks
+  while [ -h "$PRG" ] ; do
+    ls=`ls -ld "$PRG"`
+    link=`expr "$ls" : '.*-> \(.*\)$'`
+    if expr "$link" : '/.*' > /dev/null; then
+      PRG="$link"
+    else
+      PRG="`dirname "$PRG"`/$link"
+    fi
+  done
+
+  saveddir=`pwd`
+
+  M2_HOME=`dirname "$PRG"`/..
+
+  # make it fully qualified
+  M2_HOME=`cd "$M2_HOME" && pwd`
+
+  cd "$saveddir"
+  # echo Using m2 at $M2_HOME
+fi
+
+# For Cygwin, ensure paths are in UNIX format before anything is touched
+if $cygwin ; then
+  [ -n "$M2_HOME" ] &&
+    M2_HOME=`cygpath --unix "$M2_HOME"`
+  [ -n "$JAVA_HOME" ] &&
+    JAVA_HOME=`cygpath --unix "$JAVA_HOME"`
+  [ -n "$CLASSPATH" ] &&
+    CLASSPATH=`cygpath --path --unix "$CLASSPATH"`
+fi
+
+# For Mingw, ensure paths are in UNIX format before anything is touched
+if $mingw ; then
+  [ -n "$M2_HOME" ] &&
+    M2_HOME="`(cd "$M2_HOME"; pwd)`"
+  [ -n "$JAVA_HOME" ] &&
+    JAVA_HOME="`(cd "$JAVA_HOME"; pwd)`"
+fi
+
+if [ -z "$JAVA_HOME" ]; then
+  javaExecutable="`which javac`"
+  if [ -n "$javaExecutable" ] && ! [ "`expr \"$javaExecutable\" : '\([^ ]*\)'`" = "no" ]; then
+    # readlink(1) is not available as standard on Solaris 10.
+    readLink=`which readlink`
+    if [ ! `expr "$readLink" : '\([^ ]*\)'` = "no" ]; then
+      if $darwin ; then
+        javaHome="`dirname \"$javaExecutable\"`"
+        javaExecutable="`cd \"$javaHome\" && pwd -P`/javac"
+      else
+        javaExecutable="`readlink -f \"$javaExecutable\"`"
+      fi
+      javaHome="`dirname \"$javaExecutable\"`"
+      javaHome=`expr "$javaHome" : '\(.*\)/bin'`
+      JAVA_HOME="$javaHome"
+      export JAVA_HOME
+    fi
+  fi
+fi
+
+if [ -z "$JAVACMD" ] ; then
+  if [ -n "$JAVA_HOME"  ] ; then
+    if [ -x "$JAVA_HOME/jre/sh/java" ] ; then
+      # IBM's JDK on AIX uses strange locations for the executables
+      JAVACMD="$JAVA_HOME/jre/sh/java"
+    else
+      JAVACMD="$JAVA_HOME/bin/java"
+    fi
+  else
+    JAVACMD="`which java`"
+  fi
+fi
+
+if [ ! -x "$JAVACMD" ] ; then
+  echo "Error: JAVA_HOME is not defined correctly." >&2
+  echo "  We cannot execute $JAVACMD" >&2
+  exit 1
+fi
+
+if [ -z "$JAVA_HOME" ] ; then
+  echo "Warning: JAVA_HOME environment variable is not set."
+fi
+
+CLASSWORLDS_LAUNCHER=org.codehaus.plexus.classworlds.launcher.Launcher
+
+# traverses directory structure from process work directory to filesystem root
+# first directory with .mvn subdirectory is considered project base directory
+find_maven_basedir() {
+
+  if [ -z "$1" ]
+  then
+    echo "Path not specified to find_maven_basedir"
+    return 1
+  fi
+
+  basedir="$1"
+  wdir="$1"
+  while [ "$wdir" != '/' ] ; do
+    if [ -d "$wdir"/.mvn ] ; then
+      basedir=$wdir
+      break
+    fi
+    # workaround for JBEAP-8937 (on Solaris 10/Sparc)
+    if [ -d "${wdir}" ]; then
+      wdir=`cd "$wdir/.."; pwd`
+    fi
+    # end of workaround
+  done
+  echo "${basedir}"
+}
+
+# concatenates all lines of a file
+concat_lines() {
+  if [ -f "$1" ]; then
+    echo "$(tr -s '\n' ' ' < "$1")"
+  fi
+}
+
+BASE_DIR=`find_maven_basedir "$(pwd)"`
+if [ -z "$BASE_DIR" ]; then
+  exit 1;
+fi
+
+##########################################################################################
+# Extension to allow automatically downloading the maven-wrapper.jar from Maven-central
+# This allows using the maven wrapper in projects that prohibit checking in binary data.
+##########################################################################################
+if [ -r "$BASE_DIR/.mvn/wrapper/maven-wrapper.jar" ]; then
+    if [ "$MVNW_VERBOSE" = true ]; then
+      echo "Found .mvn/wrapper/maven-wrapper.jar"
+    fi
+else
+    if [ "$MVNW_VERBOSE" = true ]; then
+      echo "Couldn't find .mvn/wrapper/maven-wrapper.jar, downloading it ..."
+    fi
+    if [ -n "$MVNW_REPOURL" ]; then
+      jarUrl="$MVNW_REPOURL/io/takari/maven-wrapper/0.5.6/maven-wrapper-0.5.6.jar"
+    else
+      jarUrl="https://repo.maven.apache.org/maven2/io/takari/maven-wrapper/0.5.6/maven-wrapper-0.5.6.jar"
+    fi
+    while IFS="=" read key value; do
+      case "$key" in (wrapperUrl) jarUrl="$value"; break ;;
+      esac
+    done < "$BASE_DIR/.mvn/wrapper/maven-wrapper.properties"
+    if [ "$MVNW_VERBOSE" = true ]; then
+      echo "Downloading from: $jarUrl"
+    fi
+    wrapperJarPath="$BASE_DIR/.mvn/wrapper/maven-wrapper.jar"
+    if $cygwin; then
+      wrapperJarPath=`cygpath --path --windows "$wrapperJarPath"`
+    fi
+
+    if command -v wget > /dev/null; then
+        if [ "$MVNW_VERBOSE" = true ]; then
+          echo "Found wget ... using wget"
+        fi
+        if [ -z "$MVNW_USERNAME" ] || [ -z "$MVNW_PASSWORD" ]; then
+            wget "$jarUrl" -O "$wrapperJarPath"
+        else
+            wget --http-user=$MVNW_USERNAME --http-password=$MVNW_PASSWORD "$jarUrl" -O "$wrapperJarPath"
+        fi
+    elif command -v curl > /dev/null; then
+        if [ "$MVNW_VERBOSE" = true ]; then
+          echo "Found curl ... using curl"
+        fi
+        if [ -z "$MVNW_USERNAME" ] || [ -z "$MVNW_PASSWORD" ]; then
+            curl -o "$wrapperJarPath" "$jarUrl" -f
+        else
+            curl --user $MVNW_USERNAME:$MVNW_PASSWORD -o "$wrapperJarPath" "$jarUrl" -f
+        fi
+
+    else
+        if [ "$MVNW_VERBOSE" = true ]; then
+          echo "Falling back to using Java to download"
+        fi
+        javaClass="$BASE_DIR/.mvn/wrapper/MavenWrapperDownloader.java"
+        # For Cygwin, switch paths to Windows format before running javac
+        if $cygwin; then
+          javaClass=`cygpath --path --windows "$javaClass"`
+        fi
+        if [ -e "$javaClass" ]; then
+            if [ ! -e "$BASE_DIR/.mvn/wrapper/MavenWrapperDownloader.class" ]; then
+                if [ "$MVNW_VERBOSE" = true ]; then
+                  echo " - Compiling MavenWrapperDownloader.java ..."
+                fi
+                # Compiling the Java class
+                ("$JAVA_HOME/bin/javac" "$javaClass")
+            fi
+            if [ -e "$BASE_DIR/.mvn/wrapper/MavenWrapperDownloader.class" ]; then
+                # Running the downloader
+                if [ "$MVNW_VERBOSE" = true ]; then
+                  echo " - Running MavenWrapperDownloader.java ..."
+                fi
+                ("$JAVA_HOME/bin/java" -cp .mvn/wrapper MavenWrapperDownloader "$MAVEN_PROJECTBASEDIR")
+            fi
+        fi
+    fi
+fi
+##########################################################################################
+# End of extension
+##########################################################################################
+
+export MAVEN_PROJECTBASEDIR=${MAVEN_BASEDIR:-"$BASE_DIR"}
+if [ "$MVNW_VERBOSE" = true ]; then
+  echo $MAVEN_PROJECTBASEDIR
+fi
+MAVEN_OPTS="$(concat_lines "$MAVEN_PROJECTBASEDIR/.mvn/jvm.config") $MAVEN_OPTS"
+
+# For Cygwin, switch paths to Windows format before running java
+if $cygwin; then
+  [ -n "$M2_HOME" ] &&
+    M2_HOME=`cygpath --path --windows "$M2_HOME"`
+  [ -n "$JAVA_HOME" ] &&
+    JAVA_HOME=`cygpath --path --windows "$JAVA_HOME"`
+  [ -n "$CLASSPATH" ] &&
+    CLASSPATH=`cygpath --path --windows "$CLASSPATH"`
+  [ -n "$MAVEN_PROJECTBASEDIR" ] &&
+    MAVEN_PROJECTBASEDIR=`cygpath --path --windows "$MAVEN_PROJECTBASEDIR"`
+fi
+
+# Provide a "standardized" way to retrieve the CLI args that will
+# work with both Windows and non-Windows executions.
+MAVEN_CMD_LINE_ARGS="$MAVEN_CONFIG $@"
+export MAVEN_CMD_LINE_ARGS
+
+WRAPPER_LAUNCHER=org.apache.maven.wrapper.MavenWrapperMain
+
+exec "$JAVACMD" \
+  $MAVEN_OPTS \
+  -classpath "$MAVEN_PROJECTBASEDIR/.mvn/wrapper/maven-wrapper.jar" \
+  "-Dmaven.home=${M2_HOME}" "-Dmaven.multiModuleProjectDirectory=${MAVEN_PROJECTBASEDIR}" \
+  ${WRAPPER_LAUNCHER} $MAVEN_CONFIG "$@"

+ 182 - 0
mvnw.cmd

@@ -0,0 +1,182 @@
+@REM ----------------------------------------------------------------------------
+@REM Licensed to the Apache Software Foundation (ASF) under one
+@REM or more contributor license agreements.  See the NOTICE file
+@REM distributed with this work for additional information
+@REM regarding copyright ownership.  The ASF licenses this file
+@REM to you under the Apache License, Version 2.0 (the
+@REM "License"); you may not use this file except in compliance
+@REM with the License.  You may obtain a copy of the License at
+@REM
+@REM    https://www.apache.org/licenses/LICENSE-2.0
+@REM
+@REM Unless required by applicable law or agreed to in writing,
+@REM software distributed under the License is distributed on an
+@REM "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+@REM KIND, either express or implied.  See the License for the
+@REM specific language governing permissions and limitations
+@REM under the License.
+@REM ----------------------------------------------------------------------------
+
+@REM ----------------------------------------------------------------------------
+@REM Maven Start Up Batch script
+@REM
+@REM Required ENV vars:
+@REM JAVA_HOME - location of a JDK home dir
+@REM
+@REM Optional ENV vars
+@REM M2_HOME - location of maven2's installed home dir
+@REM MAVEN_BATCH_ECHO - set to 'on' to enable the echoing of the batch commands
+@REM MAVEN_BATCH_PAUSE - set to 'on' to wait for a keystroke before ending
+@REM MAVEN_OPTS - parameters passed to the Java VM when running Maven
+@REM     e.g. to debug Maven itself, use
+@REM set MAVEN_OPTS=-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=8000
+@REM MAVEN_SKIP_RC - flag to disable loading of mavenrc files
+@REM ----------------------------------------------------------------------------
+
+@REM Begin all REM lines with '@' in case MAVEN_BATCH_ECHO is 'on'
+@echo off
+@REM set title of command window
+title %0
+@REM enable echoing by setting MAVEN_BATCH_ECHO to 'on'
+@if "%MAVEN_BATCH_ECHO%" == "on"  echo %MAVEN_BATCH_ECHO%
+
+@REM set %HOME% to equivalent of $HOME
+if "%HOME%" == "" (set "HOME=%HOMEDRIVE%%HOMEPATH%")
+
+@REM Execute a user defined script before this one
+if not "%MAVEN_SKIP_RC%" == "" goto skipRcPre
+@REM check for pre script, once with legacy .bat ending and once with .cmd ending
+if exist "%HOME%\mavenrc_pre.bat" call "%HOME%\mavenrc_pre.bat"
+if exist "%HOME%\mavenrc_pre.cmd" call "%HOME%\mavenrc_pre.cmd"
+:skipRcPre
+
+@setlocal
+
+set ERROR_CODE=0
+
+@REM To isolate internal variables from possible post scripts, we use another setlocal
+@setlocal
+
+@REM ==== START VALIDATION ====
+if not "%JAVA_HOME%" == "" goto OkJHome
+
+echo.
+echo Error: JAVA_HOME not found in your environment. >&2
+echo Please set the JAVA_HOME variable in your environment to match the >&2
+echo location of your Java installation. >&2
+echo.
+goto error
+
+:OkJHome
+if exist "%JAVA_HOME%\bin\java.exe" goto init
+
+echo.
+echo Error: JAVA_HOME is set to an invalid directory. >&2
+echo JAVA_HOME = "%JAVA_HOME%" >&2
+echo Please set the JAVA_HOME variable in your environment to match the >&2
+echo location of your Java installation. >&2
+echo.
+goto error
+
+@REM ==== END VALIDATION ====
+
+:init
+
+@REM Find the project base dir, i.e. the directory that contains the folder ".mvn".
+@REM Fallback to current working directory if not found.
+
+set MAVEN_PROJECTBASEDIR=%MAVEN_BASEDIR%
+IF NOT "%MAVEN_PROJECTBASEDIR%"=="" goto endDetectBaseDir
+
+set EXEC_DIR=%CD%
+set WDIR=%EXEC_DIR%
+:findBaseDir
+IF EXIST "%WDIR%"\.mvn goto baseDirFound
+cd ..
+IF "%WDIR%"=="%CD%" goto baseDirNotFound
+set WDIR=%CD%
+goto findBaseDir
+
+:baseDirFound
+set MAVEN_PROJECTBASEDIR=%WDIR%
+cd "%EXEC_DIR%"
+goto endDetectBaseDir
+
+:baseDirNotFound
+set MAVEN_PROJECTBASEDIR=%EXEC_DIR%
+cd "%EXEC_DIR%"
+
+:endDetectBaseDir
+
+IF NOT EXIST "%MAVEN_PROJECTBASEDIR%\.mvn\jvm.config" goto endReadAdditionalConfig
+
+@setlocal EnableExtensions EnableDelayedExpansion
+for /F "usebackq delims=" %%a in ("%MAVEN_PROJECTBASEDIR%\.mvn\jvm.config") do set JVM_CONFIG_MAVEN_PROPS=!JVM_CONFIG_MAVEN_PROPS! %%a
+@endlocal & set JVM_CONFIG_MAVEN_PROPS=%JVM_CONFIG_MAVEN_PROPS%
+
+:endReadAdditionalConfig
+
+SET MAVEN_JAVA_EXE="%JAVA_HOME%\bin\java.exe"
+set WRAPPER_JAR="%MAVEN_PROJECTBASEDIR%\.mvn\wrapper\maven-wrapper.jar"
+set WRAPPER_LAUNCHER=org.apache.maven.wrapper.MavenWrapperMain
+
+set DOWNLOAD_URL="https://repo.maven.apache.org/maven2/io/takari/maven-wrapper/0.5.6/maven-wrapper-0.5.6.jar"
+
+FOR /F "tokens=1,2 delims==" %%A IN ("%MAVEN_PROJECTBASEDIR%\.mvn\wrapper\maven-wrapper.properties") DO (
+    IF "%%A"=="wrapperUrl" SET DOWNLOAD_URL=%%B
+)
+
+@REM Extension to allow automatically downloading the maven-wrapper.jar from Maven-central
+@REM This allows using the maven wrapper in projects that prohibit checking in binary data.
+if exist %WRAPPER_JAR% (
+    if "%MVNW_VERBOSE%" == "true" (
+        echo Found %WRAPPER_JAR%
+    )
+) else (
+    if not "%MVNW_REPOURL%" == "" (
+        SET DOWNLOAD_URL="%MVNW_REPOURL%/io/takari/maven-wrapper/0.5.6/maven-wrapper-0.5.6.jar"
+    )
+    if "%MVNW_VERBOSE%" == "true" (
+        echo Couldn't find %WRAPPER_JAR%, downloading it ...
+        echo Downloading from: %DOWNLOAD_URL%
+    )
+
+    powershell -Command "&{"^
+		"$webclient = new-object System.Net.WebClient;"^
+		"if (-not ([string]::IsNullOrEmpty('%MVNW_USERNAME%') -and [string]::IsNullOrEmpty('%MVNW_PASSWORD%'))) {"^
+		"$webclient.Credentials = new-object System.Net.NetworkCredential('%MVNW_USERNAME%', '%MVNW_PASSWORD%');"^
+		"}"^
+		"[Net.ServicePointManager]::SecurityProtocol = [Net.SecurityProtocolType]::Tls12; $webclient.DownloadFile('%DOWNLOAD_URL%', '%WRAPPER_JAR%')"^
+		"}"
+    if "%MVNW_VERBOSE%" == "true" (
+        echo Finished downloading %WRAPPER_JAR%
+    )
+)
+@REM End of extension
+
+@REM Provide a "standardized" way to retrieve the CLI args that will
+@REM work with both Windows and non-Windows executions.
+set MAVEN_CMD_LINE_ARGS=%*
+
+%MAVEN_JAVA_EXE% %JVM_CONFIG_MAVEN_PROPS% %MAVEN_OPTS% %MAVEN_DEBUG_OPTS% -classpath %WRAPPER_JAR% "-Dmaven.multiModuleProjectDirectory=%MAVEN_PROJECTBASEDIR%" %WRAPPER_LAUNCHER% %MAVEN_CONFIG% %*
+if ERRORLEVEL 1 goto error
+goto end
+
+:error
+set ERROR_CODE=1
+
+:end
+@endlocal & set ERROR_CODE=%ERROR_CODE%
+
+if not "%MAVEN_SKIP_RC%" == "" goto skipRcPost
+@REM check for post script, once with legacy .bat ending and once with .cmd ending
+if exist "%HOME%\mavenrc_post.bat" call "%HOME%\mavenrc_post.bat"
+if exist "%HOME%\mavenrc_post.cmd" call "%HOME%\mavenrc_post.cmd"
+:skipRcPost
+
+@REM pause the script if MAVEN_BATCH_PAUSE is set to 'on'
+if "%MAVEN_BATCH_PAUSE%" == "on" pause
+
+if "%MAVEN_TERMINATE_CMD%" == "on" exit %ERROR_CODE%
+
+exit /B %ERROR_CODE%

+ 192 - 0
pom.xml

@@ -0,0 +1,192 @@
+<?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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.springframework.boot</groupId>
+        <artifactId>spring-boot-starter-parent</artifactId>
+        <version>2.5.2</version>
+        <relativePath/> <!-- lookup parent from repository -->
+    </parent>
+    <groupId>com.persagy</groupId>
+    <artifactId>ztk-encryptdecodedata</artifactId>
+    <version>v0.0.1</version>
+    <name>ztk-encryptdecodedata</name>
+    <description>Cmft project data encryption service interface processing</description>
+	<packaging>jar</packaging>
+    <properties>
+        <java.version>1.8</java.version>
+    </properties>
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter</artifactId>
+        </dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-web</artifactId>
+			<!-- 移除嵌入式tomcat插件 -->
+			<!--  <exclusions>
+ 				<exclusion>
+					 <groupId>org.springframework.boot</groupId>
+ 					<artifactId>spring-boot-starter-tomcat</artifactId>
+ 				</exclusion>
+			</exclusions>  -->
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-integration</artifactId>
+			<version>2.5.2</version>
+			<exclusions>
+				<exclusion>
+					<groupId>org.springframework.boot</groupId>
+					<artifactId>spring-boot-starter-logging</artifactId>
+				</exclusion>
+			</exclusions>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.integration</groupId>
+			<artifactId>spring-integration-ip</artifactId>
+			<version>5.5.1</version>
+		</dependency>
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <optional>true</optional>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+			<exclusions>
+				<exclusion>
+					<groupId>org.junit.vintage</groupId>
+					<artifactId>junit-vintage-engine</artifactId>
+				</exclusion>
+			</exclusions>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.boot</groupId>
+			<artifactId>spring-boot-starter-aop</artifactId>
+			<version>2.5.2</version>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.integration</groupId>
+			<artifactId>spring-integration-core</artifactId>
+			<version>5.5.1</version>
+		</dependency>
+
+		<dependency>
+		<groupId>org.springframework.boot</groupId>
+		<artifactId>spring-boot-configuration-processor</artifactId>
+		<optional>true</optional>
+		</dependency>
+		<dependency>
+			<groupId>org.yaml</groupId>
+			<artifactId>snakeyaml</artifactId>
+		</dependency>
+		<dependency>
+			<groupId>com.github.ulisesbocchio</groupId>
+			<artifactId>jasypt-spring-boot-starter</artifactId>
+			<version>3.0.3</version>
+		</dependency>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-clients</artifactId>
+            <version>0.11.0.0</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-streams</artifactId>
+            <version>0.11.0.0</version>
+        </dependency>
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>fastjson</artifactId>
+            <version>1.2.54</version>
+        </dependency>
+        <dependency>
+            <groupId>commons-net</groupId>
+            <artifactId>commons-net</artifactId>
+            <version>3.3</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+            <version>3.7</version>
+        </dependency>
+        <!-- https://mvnrepository.com/artifact/cn.shuibo/rsa-encrypt-body-spring-boot -->
+        <dependency>
+            <groupId>cn.shuibo</groupId>
+            <artifactId>rsa-encrypt-body-spring-boot</artifactId>
+            <version>1.0.1.RELEASE</version>
+        </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>4.13.1</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.httpcomponents</groupId>
+            <artifactId>httpclient</artifactId>
+            <version>4.5.13</version>
+        </dependency>
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-all</artifactId>
+            <version>4.1.43.Final</version>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-maven-plugin</artifactId>
+                <configuration>
+                        <!-- 一定要配置否则打jar的时候会说找不到主类,根据具体项目来-->
+                        <mainClass>com.persagy.ztkencryptdecodedata.ZtkEncryptdecodedataApplication</mainClass>
+                        <executable>true</executable>
+                        <fork>true</fork>
+                    <excludes>
+                        <exclude>
+                            <groupId>org.projectlombok</groupId>
+                            <artifactId>lombok</artifactId>
+                        </exclude>
+                    </excludes>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+    <repositories>
+        <repository>
+            <id>spring-milestones</id>
+            <name>Spring Milestones</name>
+            <url>https://repo1.maven.org/maven2/</url>
+        </repository>
+        <repository>
+            <id>spring-snapshots</id>
+            <name>Spring Snapshots</name>
+            <url>https://repo1.maven.org/maven2/</url>
+            <snapshots>
+                <enabled>true</enabled>
+            </snapshots>
+        </repository>
+    </repositories>
+    <pluginRepositories>
+        <pluginRepository>
+            <id>spring-milestones</id>
+            <name>Spring Milestones</name>
+            <url>https://repo1.maven.org/maven2/</url>
+        </pluginRepository>
+        <pluginRepository>
+            <id>spring-snapshots</id>
+            <name>Spring Snapshots</name>
+            <url>https://repo1.maven.org/maven2/</url>
+            <snapshots>
+                <enabled>true</enabled>
+            </snapshots>
+        </pluginRepository>
+    </pluginRepositories>
+</project>

+ 26 - 0
src/main/java/com/persagy/ztkencryptdecodedata/SystemConfig.java

@@ -0,0 +1,26 @@
+package com.persagy.ztkencryptdecodedata;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.ApplicationArguments;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.core.annotation.Order;
+import org.springframework.core.env.Environment;
+import org.springframework.stereotype.Component;
+
+@Component
+@Order(value = 1)
+public class SystemConfig  implements ApplicationRunner {
+
+    @Autowired // 注入到容器中
+    private Environment environment;
+
+    @Value("${spring.location}")
+    private  String profile;
+
+
+    @Override
+    public void run(ApplicationArguments args) throws Exception {
+        System.out.println("Environment Properties Config >>> " + environment.getProperty("spring.location"));
+    }
+}

+ 23 - 0
src/main/java/com/persagy/ztkencryptdecodedata/ZtkEncryptdecodedataApplication.java

@@ -0,0 +1,23 @@
+package com.persagy.ztkencryptdecodedata;
+
+import cn.shuibo.annotation.EnableSecurity;
+import com.persagy.ztkencryptdecodedata.config.YmlPropertySourceFactory;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.web.servlet.ServletComponentScan;
+import org.springframework.context.ConfigurableApplicationContext;
+import org.springframework.context.annotation.PropertySource;
+
+
+@EnableSecurity
+@PropertySource(value = "classpath:application-${spring.profiles.active}.yml", encoding = "UTF-8",factory = YmlPropertySourceFactory.class)
+@ServletComponentScan
+@SpringBootApplication
+public class ZtkEncryptdecodedataApplication {
+
+    public static void main(String[] args) {
+        ConfigurableApplicationContext run = SpringApplication.run(ZtkEncryptdecodedataApplication.class, args);
+
+    }
+
+}

+ 27 - 0
src/main/java/com/persagy/ztkencryptdecodedata/config/Base64Util.java

@@ -0,0 +1,27 @@
+package com.persagy.ztkencryptdecodedata.config;
+
+import org.apache.commons.codec.binary.Base64;
+
+public class Base64Util{
+
+    /**
+     * Decoding to binary
+     * @param base64 base64
+     * @return byte
+     * @throws Exception Exception
+     */
+    public static byte[] decode(String base64) throws Exception {
+        return Base64.decodeBase64(base64);
+    }
+
+    /**
+     * Binary encoding as a string
+     * @param bytes byte
+     * @return String
+     * @throws Exception Exception
+     */
+    public static String encode(byte[] bytes) throws Exception {
+        return new String(Base64.encodeBase64(bytes));
+    }
+}
+

+ 100 - 0
src/main/java/com/persagy/ztkencryptdecodedata/config/RSAUtil.java

@@ -0,0 +1,100 @@
+package com.persagy.ztkencryptdecodedata.config;
+
+import cn.shuibo.util.Base64Util;
+
+import javax.crypto.Cipher;
+import java.io.ByteArrayOutputStream;
+import java.security.Key;
+import java.security.KeyFactory;
+import java.security.spec.PKCS8EncodedKeySpec;
+import java.security.spec.X509EncodedKeySpec;
+
+public class RSAUtil{
+
+    /**
+     * encryption algorithm RSA
+     */
+    public static final String KEY_ALGORITHM = "RSA";
+
+    /**
+     * RSA Maximum Encrypted Plaintext Size
+     */
+    private static final int MAX_ENCRYPT_BLOCK = 117;
+
+    /**
+     * RSA Maximum decrypted ciphertext size
+     */
+    private static final int MAX_DECRYPT_BLOCK = 256;
+
+    /**
+     * encryption
+     * @param data data
+     * @param publicKey publicKey
+     * @return byte
+     * @throws Exception Exception
+     */
+    public static byte[] encrypt(byte[] data, String publicKey)
+            throws Exception {
+        byte[] keyBytes = cn.shuibo.util.Base64Util.decode(publicKey);
+        X509EncodedKeySpec x509KeySpec = new X509EncodedKeySpec(keyBytes);
+        KeyFactory keyFactory = KeyFactory.getInstance(KEY_ALGORITHM);
+        Key publicK = keyFactory.generatePublic(x509KeySpec);
+        Cipher cipher = Cipher.getInstance(keyFactory.getAlgorithm());
+        cipher.init(Cipher.ENCRYPT_MODE, publicK);
+        int inputLen = data.length;
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        int offSet = 0;
+        byte[] cache;
+        int i = 0;
+        // Sectional Encryption of Data
+        while (inputLen - offSet > 0) {
+            if (inputLen - offSet > MAX_ENCRYPT_BLOCK) {
+                cache = cipher.doFinal(data, offSet, MAX_ENCRYPT_BLOCK);
+            } else {
+                cache = cipher.doFinal(data, offSet, inputLen - offSet);
+            }
+            out.write(cache, 0, cache.length);
+            i++;
+            offSet = i * MAX_ENCRYPT_BLOCK;
+        }
+        byte[] encryptedData = out.toByteArray();
+        out.close();
+        return encryptedData;
+    }
+
+    /**
+     * Decrypt
+     * @param text text
+     * @param privateKey privateKey
+     * @return byte
+     * @throws Exception Exception
+     */
+    public static byte[] decrypt(byte[] text, String privateKey)
+            throws Exception {
+        byte[] keyBytes = Base64Util.decode(privateKey);
+        PKCS8EncodedKeySpec pkcs8KeySpec = new PKCS8EncodedKeySpec(keyBytes);
+        KeyFactory keyFactory = KeyFactory.getInstance(KEY_ALGORITHM);
+        Key privateK = keyFactory.generatePrivate(pkcs8KeySpec);
+        Cipher cipher = Cipher.getInstance(keyFactory.getAlgorithm());
+        cipher.init(Cipher.DECRYPT_MODE, privateK);
+        int inputLen = text.length;
+        ByteArrayOutputStream out = new ByteArrayOutputStream();
+        int offSet = 0;
+        byte[] cache;
+        int i = 0;
+        // Sectional Encryption of Data
+        while (inputLen - offSet > 0) {
+            if (inputLen - offSet > MAX_DECRYPT_BLOCK) {
+                cache = cipher.doFinal(text, offSet, MAX_DECRYPT_BLOCK);
+            } else {
+                cache = cipher.doFinal(text, offSet, inputLen - offSet);
+            }
+            out.write(cache, 0, cache.length);
+            i++;
+            offSet = i * MAX_DECRYPT_BLOCK;
+        }
+        byte[] decryptedData = out.toByteArray();
+        out.close();
+        return decryptedData;
+    }
+}

+ 34 - 0
src/main/java/com/persagy/ztkencryptdecodedata/config/YmlPropertySourceFactory.java

@@ -0,0 +1,34 @@
+package com.persagy.ztkencryptdecodedata.config;
+
+import org.springframework.beans.factory.config.YamlPropertiesFactoryBean;
+import org.springframework.core.env.PropertiesPropertySource;
+import org.springframework.core.env.PropertySource;
+import org.springframework.core.io.support.DefaultPropertySourceFactory;
+import org.springframework.core.io.support.EncodedResource;
+
+import java.io.IOException;
+import java.util.Properties;
+
+public class YmlPropertySourceFactory extends DefaultPropertySourceFactory {
+
+    @Override
+    public PropertySource<?> createPropertySource(String name, EncodedResource resource) throws IOException {
+        String sourceName = name != null ? name : resource.getResource().getFilename();
+        if (!resource.getResource().exists()) {
+            return new PropertiesPropertySource(sourceName, new Properties());
+        } else if (sourceName.endsWith(".yml") || sourceName.endsWith(".yaml")) {
+            Properties propertiesFromYaml = loadYml(resource);
+            return new PropertiesPropertySource(sourceName, propertiesFromYaml);
+        } else {
+            return super.createPropertySource(name, resource);
+        }
+    }
+
+    private Properties loadYml(EncodedResource resource) throws IOException {
+        YamlPropertiesFactoryBean factory = new YamlPropertiesFactoryBean();
+        factory.setResources(resource.getResource());
+        factory.afterPropertiesSet();
+        return factory.getObject();
+    }
+
+}

+ 35 - 0
src/main/java/com/persagy/ztkencryptdecodedata/controller/TestEnDeServiceController.java

@@ -0,0 +1,35 @@
+package com.persagy.ztkencryptdecodedata.controller;
+
+import cn.shuibo.annotation.Decrypt;
+import cn.shuibo.annotation.Encrypt;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+import org.springframework.web.bind.annotation.*;
+
+@RestController
+@EnableAutoConfiguration
+public class TestEnDeServiceController {
+
+
+    /**
+     * 对返回值加密
+     * @return
+     */
+    @Encrypt
+    @GetMapping("/encryt")
+    public String test01(@RequestParam String Str ){
+
+        return Str;
+    }
+
+    /**
+     * 对传过来的加密参数进行解密
+     * @param testBean
+     * @return
+     */
+    @Decrypt
+    @PostMapping("/decrypt")
+    public String test02(@RequestBody String testBean){
+        return testBean;
+    }
+
+}

+ 21 - 0
src/main/java/com/persagy/ztkencryptdecodedata/dataSafety/DataOptRSA.java

@@ -0,0 +1,21 @@
+package com.persagy.ztkencryptdecodedata.dataSafety;
+
+import cn.shuibo.annotation.Encrypt;
+import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
+
+import java.nio.charset.StandardCharsets;
+
+
+@EnableAutoConfiguration
+public class DataOptRSA {
+
+    /**
+     * 对返回值加密
+     * @return
+     */
+    @Encrypt
+    public static String DataEncrypt(String msg){
+        msg.getBytes(StandardCharsets.UTF_8);
+        return msg;
+    }
+}

+ 6 - 0
src/main/java/com/persagy/ztkencryptdecodedata/dataSafety/DecrypInputMessageInterface.java

@@ -0,0 +1,6 @@
+package com.persagy.ztkencryptdecodedata.dataSafety;
+
+public interface DecrypInputMessageInterface {
+
+    String DecryptMsgInputMessage(String str);
+}

+ 62 - 0
src/main/java/com/persagy/ztkencryptdecodedata/dataSafety/DecryptInputMessageService.java

@@ -0,0 +1,62 @@
+package com.persagy.ztkencryptdecodedata.dataSafety;
+
+import cn.shuibo.config.SecretKeyConfig;
+import cn.shuibo.util.Base64Util;
+import cn.shuibo.util.RSAUtil;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+
+
+
+@Service
+public class DecryptInputMessageService implements  DecrypInputMessageInterface {
+
+    private Logger log = LoggerFactory.getLogger(this.getClass());
+    @Value("${rsa.encrypt.privateKey}")
+    private  String privateKey;
+
+    @Value("${rsa.encrypt.showLog}")
+    private  boolean showLog;
+
+    @Autowired
+    private SecretKeyConfig secretKeyConfig;
+
+
+    public String DecryptMsgInputMessage(String inputMessage) {
+        String charset =secretKeyConfig.getCharset();
+        if (StringUtils.isEmpty(privateKey)) {
+            throw new IllegalArgumentException("privateKey is null");
+        }
+        String content = inputMessage;
+        String decryptBody="";
+        try{
+
+        if (content.startsWith("{")) {
+            log.info("Unencrypted without decryption:{}", content);
+            decryptBody = content;
+        } else {
+            StringBuilder json = new StringBuilder();
+            content = content.replaceAll(" ", "+");
+
+            if (!StringUtils.isEmpty(content)) {
+                String[] contents = content.split("\\|");
+                for (String value : contents) {
+                    value = new String(RSAUtil.decrypt(Base64Util.decode(value), privateKey), charset);
+                    json.append(value);
+                }
+            }
+            decryptBody = json.toString();
+            if (showLog) {
+                log.info("Encrypted data received:{},After decryption:{}", content, decryptBody);
+            }
+        }}catch (Exception e){
+            e.printStackTrace();
+        }
+       return decryptBody;
+    }
+
+}

+ 6 - 0
src/main/java/com/persagy/ztkencryptdecodedata/dataSafety/EncryptInputMessageInterface.java

@@ -0,0 +1,6 @@
+package com.persagy.ztkencryptdecodedata.dataSafety;
+
+public interface EncryptInputMessageInterface {
+
+    Object beforeBodyWrite(Object body);
+}

+ 74 - 0
src/main/java/com/persagy/ztkencryptdecodedata/dataSafety/EncryptInputMessageService.java

@@ -0,0 +1,74 @@
+package com.persagy.ztkencryptdecodedata.dataSafety;
+
+
+import cn.shuibo.config.SecretKeyConfig;
+import cn.shuibo.util.Base64Util;
+import cn.shuibo.util.RSAUtil;
+import com.alibaba.fastjson.JSONObject;
+import com.alibaba.fastjson.annotation.JSONField;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+import org.springframework.util.StringUtils;
+
+@Service
+public class EncryptInputMessageService implements  EncryptInputMessageInterface{
+    private Logger log = LoggerFactory.getLogger(this.getClass());
+    @Value("${rsa.encrypt.open}")
+    private boolean encrypt;
+    @Autowired
+    private  SecretKeyConfig secretKeyConfig;
+    private static ThreadLocal<Boolean> encryptLocal = new ThreadLocal<>();
+    @Value("${rsa.encrypt.open}")
+    private  String converterType;
+
+    public boolean supports(String converterType) {
+        encrypt = false;
+        if (secretKeyConfig.isOpen()) {
+            encrypt = true;
+        }
+        return encrypt;
+    }
+     class Model {
+
+        @JSONField(jsonDirect=true)
+        public String content;
+    }
+
+
+    public  Object beforeBodyWrite(Object body) {
+        // EncryptResponseBodyAdvice.setEncryptStatus(false);
+        // Dynamic Settings Not Encrypted
+        Boolean status = encryptLocal.get();
+        if (null != status && !status) {
+            encryptLocal.remove();
+            return body;
+        }
+        if (encrypt) {
+            String publicKey = secretKeyConfig.getPublicKey();
+            try {
+                Model model =new Model();
+                model.content=body.toString();
+                String content = JSONObject.toJSONString(model);
+
+                if (!StringUtils.hasText(publicKey)) {
+                    throw new NullPointerException("Please configure rsa.encrypt.privatekeyc parameter!");
+                }
+                byte[] data = content.getBytes();
+                byte[] encodedData = RSAUtil.encrypt(data, publicKey);
+                String result = Base64Util.encode(encodedData);
+                if(secretKeyConfig.isShowLog()) {
+                    log.info("Pre-encrypted data:{},After encryption:{}", content, result);
+                }
+                return result;
+            } catch (Exception e) {
+                log.error("Encrypted data exception", e);
+            }
+        }
+        return body;
+    }
+
+
+}

+ 31 - 0
src/main/java/com/persagy/ztkencryptdecodedata/entity/ChattingUser.java

@@ -0,0 +1,31 @@
+package com.persagy.ztkencryptdecodedata.entity;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
+import lombok.Data;
+
+import java.util.concurrent.BlockingQueue;
+
+@Data
+public class ChattingUser {
+
+    public enum Type{
+        edge,
+        cloud
+    }
+    private String userId;
+
+    private ChannelHandlerContext channelHandlerContext;
+
+    private Type type;
+
+    private  int port;
+
+    private  String ip;
+
+    private WebSocketServerHandshaker handshaker;
+
+    private BlockingQueue<String> messageTcpSendQueue;
+
+    private  Object ack ;
+}

+ 20 - 0
src/main/java/com/persagy/ztkencryptdecodedata/entity/SecretKeyConfig.java

@@ -0,0 +1,20 @@
+package com.persagy.ztkencryptdecodedata.entity;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Configuration;
+
+@ConfigurationProperties(prefix = "rsa.encrypt")
+@Configuration
+@Data
+public class SecretKeyConfig {
+    private String privateKey;
+
+    private String publicKey;
+
+    private String charset = "UTF-8";
+
+    private boolean open = true;
+
+    private boolean showLog = false;
+}

+ 139 - 0
src/main/java/com/persagy/ztkencryptdecodedata/kafka/CloudKafkaConsumerFromEdgeTopic.java

@@ -0,0 +1,139 @@
+package com.persagy.ztkencryptdecodedata.kafka;
+
+import com.alibaba.fastjson.JSONObject;
+import com.persagy.ztkencryptdecodedata.dataSafety.DecryptInputMessageService;
+import com.persagy.ztkencryptdecodedata.netty.NettyClinet;
+import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.TopicPartition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.CommandLineRunner;
+import org.springframework.core.annotation.Order;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+import java.util.Collection;
+import java.util.Properties;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+
+@Component
+@Order(value = 2)
+public class CloudKafkaConsumerFromEdgeTopic implements CommandLineRunner {
+
+    private  static Logger logger = LoggerFactory.getLogger(CloudKafkaConsumerFromEdgeTopic.class.getClass());
+    BlockingQueue<String> messageTcpSendQueue = new LinkedBlockingQueue<>(1024 * 1024);
+     //   BlockingQueue messageTcpSendQueue =null;
+
+    @Autowired
+    private DecryptInputMessageService decryptInputMessageService;
+
+    @Autowired
+    private NettyClinet nettyClinet;
+
+    @Value("${spring.kafka.consumer.bootstrap-servers}")
+    private  String consumerBootstrap;
+    @Value("${spring.kafka.consumer.group-id}")
+    private  String groupId;
+    @Value("${spring.kafka.consumer.auto-offset-reset}")
+    private  String autooffet;
+    @Value("${spring.kafka.consumer.fetch-min-size}")
+    private  int  fetchMinSize;
+    @Value("${spring.kafka.consumer.fetch-max-wait}")
+    private  int  fetchMaxWait;
+    @Value("${spring.kafka.consumer.enable-auto-commit}")
+    private  boolean enableAutoCommit;
+    @Value("${spring.kafka.consumer.max-poll-records}")
+    private  int maxPollRecords;
+    @Value("${spring.kafka.consumer.key-deserializer}")
+    private  String keyDeserializer;
+    @Value("${spring.kafka.consumer.value-deserializer}")
+    private String valueDeserializer;
+
+    @Value("${spring.location}")
+    private String location;
+
+
+
+    @PostConstruct
+    public void init() {
+        nettyClinet.setMessageTcpSendQueue(messageTcpSendQueue);
+    }
+
+    @Override
+    public void run(String... args) throws Exception {
+        int i=0;
+        System.out.println("Colud_Subscribe_Edge_Update_messages");
+        Properties props = new Properties();
+        props.put("bootstrap.servers", consumerBootstrap);
+        props.put("group.id", groupId);
+        props.put("auto.offset.reset", autooffet);
+        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
+        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
+
+        //metadata.max.age.ms 1000
+        KafkaConsumer<String, String> consumer =  new KafkaConsumer<>(props);
+
+        String fixTopic="";
+
+        if ("Cloud".equals(location)){
+            fixTopic="Edge";
+        }
+        if ("Edge".equals(location)){
+            fixTopic ="Cloud";
+        }
+
+        Pattern pattern = Pattern.compile(fixTopic+".*");
+        consumer.subscribe(pattern, new ConsumerRebalanceListener() {
+
+            @Override
+            public void onPartitionsRevoked(Collection<TopicPartition> arg0) {
+                // TODO nothing:再均衡监听器
+            }
+
+            @Override
+            public void onPartitionsAssigned(Collection<TopicPartition> arg0) {
+                // TODO nothing:再均衡监听器
+            }
+        });
+        try {
+
+        while (true) {
+            /*
+             * poll() 方法返回一个记录列表。
+             * 每条记录都包含了记录所属主题的信息、记录所在分区的信息、记录在分区里的偏移量,以及记录的键值对。
+             * 我们一般会遍历这个列表,逐条处理这些记录。
+             * 传给poll() 方法的参数是一个超时时间,用于控制 poll() 方法的阻塞时间(在消费者的缓冲区里没有可用数据时会发生阻塞)。
+             * 如果该参数被设为 0,poll() 会立即返回,否则它会在指定的毫秒数内一直等待 broker 返回数据。
+             * 而在经过了指定的时间后,即使还是没有获取到数据,poll()也会返回结果。
+             */
+            ConsumerRecords<String, String> records = consumer.poll(1000);
+            for (ConsumerRecord<String, String> record : records) {
+                System.out.println(">>>>>>>"+i);
+                System.out.println("topic = " + record.topic() + ", partition = " + record.partition()+", messages = "+ JSONObject.parseObject(decryptInputMessageService.DecryptMsgInputMessage(record.value())).toString());
+                messageTcpSendQueue.offer(JSONObject.parseObject(decryptInputMessageService.DecryptMsgInputMessage(record.value())).toString(), 1000, TimeUnit.MICROSECONDS);
+               // messageTcpSendQueue.add(JSONObject.parseObject(decryptInputMessageService.DecryptMsgInputMessage(record.value())).toString());
+
+            }
+
+        }
+
+        } finally {
+
+        /*
+         * 在退出应用程序之前使用 close() 方法关闭消费者。
+         * 网络连接和 socket 也会随之关闭,并立即触发一次再均衡,而不是等待群组协调器发现它不再发送心跳并认定它已死亡,
+         * 因为那样需要更长的时间,导致整个群组在一段时间内无法读取消息。
+         */
+        consumer.close();
+    }
+    }
+
+}

+ 132 - 0
src/main/java/com/persagy/ztkencryptdecodedata/kafka/EdgeKafkaProducer.java

@@ -0,0 +1,132 @@
+package com.persagy.ztkencryptdecodedata.kafka;
+
+import com.alibaba.fastjson.JSONObject;
+import com.persagy.ztkencryptdecodedata.dataSafety.EncryptInputMessageService;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.errors.AuthorizationException;
+import org.apache.kafka.common.errors.OutOfOrderSequenceException;
+import org.apache.kafka.common.errors.ProducerFencedException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+
+import java.util.Properties;
+import java.util.concurrent.*;
+
+@Service
+public class EdgeKafkaProducer {
+
+    private  static  Logger logger = LoggerFactory.getLogger(EdgeKafkaProducer.class.getClass());
+    private static BlockingQueue messageQueue = null;
+
+    private static Properties properties;
+    private static KafkaProducer<String, String> producer;
+
+    @Autowired
+    private EncryptInputMessageService encryptInputMessageService;
+
+    @Value("${spring.kafka.bootstrap-servers}")
+    private  String bootstrapServer;
+    @Value("${spring.kafka.producer.acks}")
+    private  String ackSet;
+    @Value("${spring.kafka.producer.retries}")
+    private  int retries;
+    @Value("${spring.kafka.producer.batch-size}")
+    private  int batchSize;
+    @Value("${spring.kafka.producer.buffer-memory}")
+    private   int bfferMemory;
+    @Value("${spring.kafka.producer.client-id}")
+    private   String clientID;
+    @Value("${spring.kafka.producer.key-serializer}")
+    private  String keySerializer;
+    @Value("${spring.kafka.producer.value-serializer}")
+    private  String ValueSerializer;
+
+    @Value("${spring.location}")
+    private String location;
+
+
+    public  Future ProducerSend(String moudle,Object object) {
+        properties = new Properties();
+        // 连接的 kafka 集群地址
+        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer);
+        properties.put("acks", ackSet);//所有follower都响应了才认为消息提交成功,即"committed"
+        properties.put("retries", retries);//retries = MAX 无限重试,直到你意识到出现了问题:)
+        properties.put("batch.size", batchSize);//producer将试图批处理消息记录,以减少请求次数.默认的批量处理消息字节数
+        properties.put("max.request.size", 1048576); //信息发送最大值1MB
+        //batch.size当批量的数据大小达到设定值后,就会立即发送,不顾下面的linger.ms
+        properties.put("retry.backoff.ms", 500);//设定重试时间间隔避免无效的频繁重试
+        properties.put("client.id", clientID);//设定重试时间间隔避免无效的频繁重试
+        properties.put("linger.ms", 5);//延迟1ms发送,这项设置将通过增加小的延迟来完成--即,不是立即发送一条记录,producer将会等待给定的延迟时间以允许其他消息记录发送,这些消息记录可以批量处理
+        properties.put("buffer.memory", bfferMemory);//producer可以用来缓存数据的内存大小。
+        properties.put("key.serializer", "org.apache.kafka.common.serialization.IntegerSerializer");
+        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
+        // producer = new KafkaProducer<String, String>(properties);
+        producer = new KafkaProducer<String, String>(properties);
+        String Topic=location+"_"+moudle;
+
+        String message = object.toString();
+        ProducerRecord<String, String> record = new ProducerRecord<>(Topic, message);//Topic Key Value
+        // ProducerRecord<String, String> record = new ProducerRecord<>("BBF464_IoT_Edge_Test_1", message);//Topic Key Value
+        Future future = producer.send(record);
+        producer.close();
+        return future;
+    }
+    public void setQueue(BlockingQueue<String> queue) {
+        messageQueue = queue;
+        ExecutorService executorService = new ThreadPoolExecutor(10,20, 0L, TimeUnit.MILLISECONDS,
+                new LinkedBlockingQueue<>());
+        executorService.submit((Runnable) () -> {
+            while (true) {
+                try {
+                    /// Thread.sleep(3000);
+                    Object messageObject = messageQueue.peek();
+                    if (null == messageObject) {
+                        // System.out.println(">>>> queue has no data");
+                        logger.info(">>> kafka queue size: " + messageQueue.size());
+                        Thread.sleep(1000);
+
+                    } else {
+
+                        try {
+                            JSONObject object =JSONObject.parseObject(messageObject.toString());
+                            Future future = ProducerSend( object.getString("port")+"_"+object.getString("userid"),encryptInputMessageService.beforeBodyWrite(messageObject));
+                            //   System.out.println("kafkafuture=="+future.isDone());
+                            logger.info(">>> kafka status future: " + future.isDone());
+                            //  System.out.println(">>> kafka queue size: " + messageQueue.size());
+                            logger.info(">>> kafka queue size: " + messageQueue.size());
+                            if (future.isDone()) {
+                                logger.info(">>> Edge send  success msg to Cloud: " + messageObject.toString());
+                                logger.info(">>> send success encryptmsg: " + encryptInputMessageService.beforeBodyWrite(messageObject).toString());
+                                //  System.out.println("send success msg=== "+messageObject.toString());
+                                messageQueue.poll();
+                            }
+                            // System.out.println("-----------------");
+                        } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
+                            e.printStackTrace();
+                            logger.error("--- kafka_send_error:"+e);
+                        } catch (KafkaException e) {
+                            // For all other exceptions, just abort the transaction and try again.
+                            //  producer.abortTransaction();
+                            e.printStackTrace();
+                            logger.error("--- kafka_client_error:"+e);
+                        }
+                        //  logger.info(">>> kafka message: " + messageObject.toString());
+                        Thread.sleep(1000 );
+                    }
+                } catch (InterruptedException e) {
+                    logger.error("executorService_error" + e);
+                } catch (Exception e){
+                    e.printStackTrace();
+                    logger.error("executorService_error" + e);
+                }
+            }
+        });
+       // executorService.shutdown();
+    }
+}

+ 7 - 0
src/main/java/com/persagy/ztkencryptdecodedata/netty/NettyClientInterface.java

@@ -0,0 +1,7 @@
+package com.persagy.ztkencryptdecodedata.netty;
+
+import java.util.concurrent.BlockingQueue;
+
+public interface NettyClientInterface {
+    public  void  setMessageTcpSendQueue(BlockingQueue<String> messageTcpSendQueue);
+}

+ 172 - 0
src/main/java/com/persagy/ztkencryptdecodedata/netty/NettyClinet.java

@@ -0,0 +1,172 @@
+package com.persagy.ztkencryptdecodedata.netty;
+
+import com.alibaba.fastjson.JSONObject;
+import com.persagy.ztkencryptdecodedata.entity.ChattingUser;
+import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.*;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.handler.codec.string.StringDecoder;
+import io.netty.handler.codec.string.StringEncoder;
+import io.netty.util.CharsetUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+import javax.annotation.PostConstruct;
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.concurrent.*;
+
+
+@Component
+public class NettyClinet implements NettyClientInterface {
+
+    private  static Logger logger = LoggerFactory.getLogger(NettyClinet.class.getClass());
+   // private static BlockingQueue messageTcpSendQueue = new LinkedBlockingQueue<>(1024 * 1024);
+   BlockingQueue<String> messageTcpSendQueue = new LinkedBlockingQueue<>(1024 * 1024);
+
+
+    @PostConstruct
+    public void init() {
+       setMessageTcpSendQueue(messageTcpSendQueue);
+    }
+
+    static class  GlobeTime {
+        static public  String  globetime ="0";
+
+        static public  Object ackstatue;
+
+        static  public  int  globecount =0;
+
+        public static  void BuildAckstatue(Object ackstatue1){
+            ackstatue=ackstatue1;
+        }
+
+        public static  void BuildValue(String currentTime){
+            globetime=currentTime;
+
+        }
+        public static int BuilValue(int count){
+            globecount=count;
+            return globecount;
+        }
+    }
+
+    /*
+     * 将时间转换为时间戳
+     */
+    public static String dateToStamp(String s) throws ParseException {
+        String res;
+         SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss:SSS");//格式化
+        Date date = sdf.parse(s);
+        long ts = date.getTime();
+        res = String.valueOf(ts);
+        return res;
+    }
+
+
+        public void  setMessageTcpSendQueue(BlockingQueue<String> messageTcpSendQueue){
+
+            int poolSize = 1;
+
+            ExecutorService executorService =  new ThreadPoolExecutor(poolSize,poolSize, 0L, TimeUnit.MILLISECONDS,
+                    new LinkedBlockingQueue<>());
+
+            executorService.submit(() -> {
+
+                while (true) {
+                    try {
+
+                            while (!messageTcpSendQueue.isEmpty()) {
+                                /// Thread.sleep(3000);
+                                Bootstrap bootstrap = new Bootstrap();
+                                EventLoopGroup group = new NioEventLoopGroup();
+                                try {
+                                    bootstrap.group(group).channel(NioSocketChannel.class);
+                                    bootstrap.handler(new ChannelInitializer<Channel>() {
+                                        @Override
+                                        protected void initChannel(Channel ch) throws Exception {
+                                            ChannelPipeline pipeline = ch.pipeline();
+                                            //  pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
+                                            //  pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
+                                            pipeline.addLast("decoder", new StringDecoder(CharsetUtil.UTF_8));
+                                            pipeline.addLast("encoder", new StringEncoder(CharsetUtil.UTF_8));
+                                            pipeline.addLast("handler", new TcpClientHandler());
+                                            //pipeline.addLast("log", new LoggingHandler(LogLevel.INFO));
+                                        }
+                                    });
+                                    logger.info("count1>>>>>>" + GlobeTime.BuilValue(GlobeTime.globecount + 1));
+                                    Object messageObject = messageTcpSendQueue.peek();
+                                    if (null == messageObject) {
+                                        // System.out.println(">>>> queue has no data");
+                                        logger.info(">>> Tcp queue size: " + messageTcpSendQueue.size());
+                                        Thread.sleep(1000);
+
+                                    } else {
+                                        JSONObject object = JSONObject.parseObject(messageObject.toString());
+                                        bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
+                                        // Start the client.
+                                        String currentTime = dateToStamp(object.getJSONObject("content").getString("time"));
+
+                                        if (!(GlobeTime.globetime).equals(currentTime)) {
+                                            ChannelFuture future = bootstrap.connect(object.getJSONObject("content").getString("targetAddress"), Integer.valueOf(object.getJSONObject("content").getString("port"))).sync();
+                                            try {
+
+                                                logger.info(">>> Tcp queue size: " + messageTcpSendQueue.size());
+                                                logger.info("连接建立!" + future.channel().id());
+                                                future.channel().writeAndFlush(object.getJSONObject("content").getString("msg")).sync();
+
+                                                logger.info("sent——to——tcpserver>>>" + object.getJSONObject("content").getString("msg"));
+                                                TimeUnit.SECONDS.sleep(1);
+
+                                                if (future.isDone()) {
+                                                    while (true) {
+                                                        ChattingUser info = new ChattingUser();
+                                                        if ("ACK!".equals(GlobeTime.ackstatue)) {
+                                                            messageTcpSendQueue.poll();
+                                                            info.setAck("false");
+                                                            GlobeTime.BuildValue(currentTime);
+                                                            GlobeTime.BuildAckstatue(null);
+                                                            GlobeTime.BuilValue(0);
+                                                            break;
+                                                        } else {
+                                                            continue;
+                                                        }
+
+                                                    }
+
+                                                }
+
+                                                future.channel().closeFuture().sync();
+                                                // System.out.println("-----------------");
+
+                                            } catch (Exception e) {
+                                                e.printStackTrace();
+                                                logger.info(">>> 连接未建立>>>Tcp queue size: " + messageTcpSendQueue.size());
+                                                logger.info(object.getJSONObject("content").getString("targetAddress") + ":" + Integer.valueOf(object.getJSONObject("content").getString("port")) + "连接未建立!");
+                                                logger.error("--- tcp_send_error:" + e);
+                                            } finally {
+                                                future.channel().closeFuture().sync();
+                                            }
+                                        }
+                                    }
+                                    //  logger.info(">>> kafka message: " + messageObject.toString());
+                                    Thread.sleep(1000 * 5);
+
+                                } catch (InterruptedException e) {
+                                    logger.error("--- tcp_send_error:" + e);
+                                    e.printStackTrace();
+                                } finally {
+                            group.shutdownGracefully();
+                        }
+                }}catch (Exception e){
+                    e.printStackTrace();}
+                }
+
+
+            });
+             executorService.shutdown();
+        }
+
+}

+ 8 - 0
src/main/java/com/persagy/ztkencryptdecodedata/netty/NettyConstants.java

@@ -0,0 +1,8 @@
+package com.persagy.ztkencryptdecodedata.netty;
+
+public class NettyConstants {
+    public static final int SERVER_READ_IDEL_TIME_OUT = 3;
+    public static final int SERVER_WRITE_IDEL_TIME_OUT = 3;
+
+    public static final int SERVER_ALL_IDEL_TIME_OUT = 6;
+}

+ 235 - 0
src/main/java/com/persagy/ztkencryptdecodedata/netty/NettyServer.java

@@ -0,0 +1,235 @@
+package com.persagy.ztkencryptdecodedata.netty;
+
+import com.alibaba.fastjson.JSONObject;
+import com.persagy.ztkencryptdecodedata.dataSafety.DecryptInputMessageService;
+import com.persagy.ztkencryptdecodedata.dataSafety.EncryptInputMessageService;
+import com.persagy.ztkencryptdecodedata.entity.ChattingUser;
+import com.persagy.ztkencryptdecodedata.kafka.EdgeKafkaProducer;
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.*;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.util.CharsetUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+import javax.annotation.PostConstruct;
+import java.net.InetSocketAddress;
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+@Component
+public class NettyServer {
+
+    private Logger logger = LoggerFactory.getLogger(this.getClass());
+    //private static BlockingQueue messageQueue = null;
+    BlockingQueue<String> messageQueue = new LinkedBlockingQueue<>(1024 * 1024);
+    private SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss:SSS");//格式化
+
+    @Autowired
+    private EncryptInputMessageService encryptInputMessageService;
+
+    @Autowired
+    private DecryptInputMessageService decryptInputMessageService;
+
+    @Autowired
+    private EdgeKafkaProducer edgeKafkaProducer;
+
+
+    @Value("${listener.port}")
+    private String listenerPort;
+
+    private ChannelFuture [] ChannelFutures = null;
+    private List<ChattingUser> list = new ArrayList<>();
+
+    public String GetRemoteIP(String str){
+        String result ="";
+        String regEx="((2[0-4]\\d|25[0-5]|[01]?\\d\\d?)\\.){3}(2[0-4]\\d|25[0-5]|[01]?\\d\\d?)";
+        Pattern p = Pattern.compile(regEx);
+        Matcher m = p.matcher(str);
+        while (m.find()) {
+            result = m.group();
+           // System.out.println(result);
+            //break;   加break则提取string中的一个IP
+        }
+        return result;
+    }
+
+
+
+
+    public static String convertHexToString(String hex) {
+        StringBuilder sb = new StringBuilder();
+        StringBuilder sb2 = new StringBuilder();
+        for (int i = 0; i < hex.length() - 1; i += 2) {
+            String s = hex.substring(i, (i + 2));
+            int decimal = Integer.parseInt(s, 16);
+            sb.append((char) decimal);
+            sb2.append(decimal);
+        }
+        return sb.toString();
+    }
+    @PostConstruct
+    private void start() {
+
+        EventLoopGroup bossGroup = new NioEventLoopGroup();
+        NioEventLoopGroup workerGroup = new NioEventLoopGroup();
+        ServerBootstrap serverBootstrap = new ServerBootstrap();
+        edgeKafkaProducer.setQueue(messageQueue);
+        serverBootstrap.group(bossGroup,workerGroup);
+        serverBootstrap.channel(NioServerSocketChannel.class);
+        serverBootstrap.childOption(ChannelOption.SO_REUSEADDR,true);
+
+        serverBootstrap.childHandler(new ChannelInitializer() {
+            @Override
+            protected void initChannel(Channel ch) throws Exception {
+                ch.pipeline().addLast(new CountHandler());
+            }
+        });
+
+        if (ChannelFutures == null){
+            ChannelFutures = new ChannelFuture[8];
+        }
+
+
+
+        //多端口绑定
+        String[] ipPortArr =listenerPort.split(",");
+        for (int i = 1;i <= ipPortArr.length ;i++){
+            ChattingUser chattingUser = new ChattingUser();
+            chattingUser.setUserId(ipPortArr[i-1].split(":")[1]);
+            chattingUser.setIp(ipPortArr[i-1].split(":")[2]);
+            int port = Integer.valueOf(ipPortArr[i-1].split(":")[3]);
+            chattingUser.setPort(port);
+            list.add(chattingUser);
+
+            ChannelFuture channelFuture = serverBootstrap.bind(port);
+            ChannelFutures[i] = channelFuture;
+            channelFuture.addListener(future -> {
+                ;
+                if (future.isSuccess()) {
+                    System.out.println("Started success,port:" + port+">>>"+ channelFuture.channel().pipeline().channel().id().asLongText());
+                } else {
+                    System.out.println("Started Failed,port:" + port +">>>"+ channelFuture.channel().pipeline().channel().id().asLongText());
+                }
+            });
+
+        }
+    }
+
+
+
+    @ChannelHandler.Sharable
+    class CountHandler extends ChannelInboundHandlerAdapter {
+
+//        private AtomicInteger nConnection = new AtomicInteger();
+//
+//        public CountHandler() {
+//            Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> {
+//                System.out.println("connections: " + nConnection.get());
+//            }, 0, 2, TimeUnit.SECONDS);
+//
+//        }
+
+        @Override
+        public void channelActive(ChannelHandlerContext ctx) throws Exception {
+            //nConnection.incrementAndGet();
+           // System.out.println(">>>>>"+nConnection.incrementAndGet());
+            InetSocketAddress inteSocket = (InetSocketAddress) ctx.channel().remoteAddress();
+
+            System.out.println("端口 已有客户端连接***"+inteSocket.getPort());
+
+            super.channelActive(ctx);
+        }
+
+
+
+        @Override
+        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+            //super.channelRead(ctx, msg);
+           // nConnection.decrementAndGet();
+            InetSocketAddress inteSocket = (InetSocketAddress) ctx.channel().localAddress();
+            String localip = inteSocket.getAddress().getHostAddress();
+            String localport = String.valueOf(inteSocket.getPort());
+            System.out.println(
+                    "server ip: " + localip + "Server port "+ localport
+            );
+
+           // System.out.println(">>>>>"+nConnection.decrementAndGet());
+            String time = sdf.format(new Date());
+            ByteBuf in = (ByteBuf) msg;
+            System.out.println(
+                    "server receive: " + in.toString(CharsetUtil.UTF_8)
+            );
+            String str = in.toString(CharsetUtil.UTF_8);
+            JSONObject obj =new JSONObject();
+            obj.put("port",localport);
+            obj.put("msg",str);
+            obj.put("time",time);
+
+             //EncryptInputMessage.beforeBodyWrite(obj);
+            for (int j=0;j<list.size();j++){
+                if (Integer.valueOf(localport) == list.get(j).getPort() ){
+                    obj.put("userid",list.get(j).getUserId());
+                    obj.put("targetAddress",list.get(j).getIp());
+
+                }
+            }
+
+            messageQueue.offer(obj.toString(), 1000, TimeUnit.MICROSECONDS);
+
+            //System.out.println("dataEncrypt>>>"+encryptInputMessageService.beforeBodyWrite(obj).toString());
+           // decryptInputMessageService.DecryptMsgInputMessage(encryptInputMessageService.beforeBodyWrite(obj).toString());
+            //System.out.println("dataDecrypt>>>"+ decryptInputMessageService.DecryptMsgInputMessage(encryptInputMessageService.beforeBodyWrite(obj).toString()));
+
+
+            /**
+             *
+             * ctx.write(in);
+             * Bytebuf并不适合直接重用,必须新建才能写回客户端,
+             * 否则将会报io.netty.util.IllegalReferenceCountException: refCnt: 0, decrement: 1
+             * 原因是这是因为Netty4有引用计数器的原因,自从Netty 4开始,对象的生命周期由它们的引用计数(reference counts)管理,
+             * 而不是由垃圾收集器(garbage collector)管理了。ByteBuf是最值得注意的,它使用了引用计数来改进分配内存和释放内存的性能。
+             * 在我们创建ByteBuf对象后,它的引用计数是1,当你释放(release)引用计数对象时,它的引用计数减1,
+             * 如果引用计数为0,这个引用计数对象会被释放(deallocate),并返回对象池。
+             * 当尝试访问引用计数为0的引用计数对象会抛出IllegalReferenceCountException异常:
+             * 或者在该Bytebuf 在复用前需要调用retain(),将计数器置为1
+             */
+           // ByteBuf pong = Unpooled.copiedBuffer(str.getBytes());
+            //ctx.write(pong);
+
+           System.out.println(GetRemoteIP(String.valueOf(ctx.channel().remoteAddress()))+"接受到..Server received: " + in.toString(CharsetUtil.UTF_8));
+           Channel channel =ctx.channel();
+          ;
+            //System.out.println( channel.pipeline().get("id"));
+
+           String ip = GetRemoteIP(String.valueOf(ctx.channel().remoteAddress()));
+            if (list.contains(ip)) {
+                System.out.println("1111");
+            }else {
+                System.out.println("2222>>>>"+ channel.id().asLongText());
+            }
+
+           // ctx.write(in);
+            //ctx.write(msg);
+           // ctx.flush();
+
+        }
+        @Override
+        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+            cause.printStackTrace();
+            ctx.close();
+        }
+
+    }
+}

+ 19 - 0
src/main/java/com/persagy/ztkencryptdecodedata/netty/TcpClientHandler.java

@@ -0,0 +1,19 @@
+package com.persagy.ztkencryptdecodedata.netty;
+
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TcpClientHandler extends SimpleChannelInboundHandler<Object> {
+    private static Logger logger = LoggerFactory.getLogger(NettyClinet.class);
+
+    @Override
+    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
+
+        NettyClinet.GlobeTime.BuilValue(NettyClinet.GlobeTime.globecount+1);
+        NettyClinet.GlobeTime.BuildAckstatue(msg);
+        logger.info("client接收到服务器返回的消息>>>" + msg );
+    }
+}

文件差異過大導致無法顯示
+ 33 - 0
src/main/resources/application-dev.yml


文件差異過大導致無法顯示
+ 33 - 0
src/main/resources/application-prod.yml


文件差異過大導致無法顯示
+ 33 - 0
src/main/resources/application-uat.yml


+ 7 - 0
src/main/resources/application.yml

@@ -0,0 +1,7 @@
+spring:
+  profiles:
+    active: dev
+  location: Edge  # Edge 边缘测 or Cloud 云端
+
+
+

+ 57 - 0
src/test/java/com/persagy/ztkencryptdecodedata/TcpServer.java

@@ -0,0 +1,57 @@
+package com.persagy.ztkencryptdecodedata;
+
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.handler.codec.string.StringDecoder;
+import io.netty.handler.codec.string.StringEncoder;
+import io.netty.util.CharsetUtil;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TcpServer {
+    private static final int PORT = 30092;
+    private static Logger logger = LoggerFactory.getLogger(TcpServer.class);
+
+    protected static void start() {
+        final EventLoopGroup bossGroup = new NioEventLoopGroup(2);
+        final EventLoopGroup workerGroup = new NioEventLoopGroup(4);
+        try {
+            ServerBootstrap b = new ServerBootstrap();
+            b.group(bossGroup, workerGroup);
+            b.channel(NioServerSocketChannel.class);
+            b.childHandler(new ChannelInitializer<SocketChannel>() {
+                @Override
+                public void initChannel(SocketChannel ch) throws Exception {
+                    ChannelPipeline pipeline = ch.pipeline();
+                    //pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
+                   // pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
+                    pipeline.addLast("decoder", new StringDecoder(CharsetUtil.UTF_8));
+                    pipeline.addLast("encoder", new StringEncoder(CharsetUtil.UTF_8));
+                    pipeline.addLast(new TcpServerHandler());
+                }
+            });
+
+            // Start the server.
+            ChannelFuture f = b.bind(PORT).sync();
+            logger.info("TCP服务器已启动");
+
+            // Wait until the server socket is closed.
+            f.channel().closeFuture().sync();
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        } finally {
+            workerGroup.shutdownGracefully();
+            bossGroup.shutdownGracefully();
+        }
+    }
+
+    public static void main(String[] args) throws Exception {
+        TcpServer.start();
+    }
+}

+ 23 - 0
src/test/java/com/persagy/ztkencryptdecodedata/TcpServerHandler.java

@@ -0,0 +1,23 @@
+package com.persagy.ztkencryptdecodedata;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TcpServerHandler extends SimpleChannelInboundHandler<Object> {
+
+    private static Logger logger = LoggerFactory.getLogger(TcpServerHandler.class);
+
+    @Override
+    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
+        ctx.channel().writeAndFlush("server收到消息:" + msg);
+    }
+
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+        logger.warn("exceptionCaught!", cause);
+        ctx.close();
+    }
+
+}

+ 13 - 0
src/test/java/com/persagy/ztkencryptdecodedata/ZtkEncryptdecodedataApplicationTests.java

@@ -0,0 +1,13 @@
+package com.persagy.ztkencryptdecodedata;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+
+@SpringBootTest
+class ZtkEncryptdecodedataApplicationTests {
+
+    @Test
+    void contextLoads() {
+    }
+
+}