xml复制代码        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-aop</artifactId>
        </dependency>

typescript复制代码/**
 * Controller层的日志封装
 * @author Summerday
 */
@Data
@ToString
public class WebLog implements Serializable {

    private static final long serialVersionUID = 1L;

    // 操作描述
    private String description;

    // 操作时间
    private Long startTime;

    // 消耗时间
    private Integer timeCost;

    // URL
    private String url;

    // URI
    private String uri;

    // 请求类型
    private String httpMethod;

    // IP地址
    private String ipAddress;

    // 请求参数
    private Object params;

    // 请求返回的结果
    private Object result;

    // 操作类型
    private String methodType;
}

less复制代码@Target({ElementType.PARAMETER,ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface Log {

    /**
     * 描述
     */
    String description() default "";

    /**
     * 方法类型 INSERT DELETE UPDATE OTHER
     */
    MethodType methodType() default MethodType.OTHER;
}

less复制代码@RestController
public class HelloController {

    @PostMapping("/hello")
    @Log(description = "hello post",methodType = MethodType.INSERT)
    public String hello(@RequestBody User user) {
        return "hello";
    }

    @GetMapping("/hello")
    @Log(description = "hello get")
    public String hello(@RequestParam("name") String username, String hobby) {
        int a = 1 / 0;
        return "hello";
    }
}

java复制代码/**
 * 定义切面
 * @author Summerday
 */

@Aspect
@Component
public class LogAspect {

    private static final Logger log = LoggerFactory.getLogger(LogAspect.class);

    /**
     * web层切点
     * 1. @Pointcut("execution(public * com.hyh.web.*.*(..))")  web层的所有方法
     * 2. @Pointcut("@annotation(com.hyh.annotation.Log)")      Log注解标注的方法
     */
    @Pointcut("@annotation(com.hyh.annotation.Log)")
    public void webLog() {
    }
}

scss复制代码/**
 * 定义切面
 * @author Summerday
 */

@Aspect
@Component
public class LogAspect {

    private static final Logger log = LoggerFactory.getLogger(LogAspect.class);

    /**
     * web层切点
     * 1. @Pointcut("execution(public * com.hyh.web.*.*(..))")  web层的所有方法
     * 2. @Pointcut("@annotation(com.hyh.annotation.Log)")      Log注解标注的方法
     */

    @Pointcut("@annotation(com.hyh.annotation.Log)")
    public void webLog() {
    }


    /**
     * 环绕通知
     */
    @Around("webLog()")
    public Object doAround(ProceedingJoinPoint joinPoint) throws Throwable {
        //获取请求对象
        HttpServletRequest request = getRequest();
        WebLog webLog = new WebLog();
        Object result = null;
        try {
            log.info("=================前置通知=====================");
            long start = System.currentTimeMillis();
            result = joinPoint.proceed();
            log.info("=================返回通知=====================");
            long timeCost = System.currentTimeMillis() - start;
            // 获取Log注解
            Log logAnnotation = getAnnotation(joinPoint);
            // 封装webLog对象
            webLog.setMethodType(logAnnotation.methodType().name());
            webLog.setDescription(logAnnotation.description());
            webLog.setTimeCost((int) timeCost);
            webLog.setStartTime(start);
            webLog.setIpAddress(request.getRemoteAddr());
            webLog.setHttpMethod(request.getMethod());
            webLog.setParams(getParams(joinPoint));
            webLog.setResult(result);
            webLog.setUri(request.getRequestURI());
            webLog.setUrl(request.getRequestURL().toString());
            log.info("{}", JSONUtil.parse(webLog));
        } catch (Throwable e) {
            log.info("==================异常通知=====================");
            log.error(e.getMessage());
            throw new Throwable(e);
        }finally {
            log.info("=================后置通知=====================");
        }
        return result;
    }

    /**
     * 获取方法上的注解
     */
    private Log getAnnotation(ProceedingJoinPoint joinPoint) {
        Method method = ((MethodSignature) joinPoint.getSignature()).getMethod();
        return method.getAnnotation(Log.class);
    }


    /**
     * 获取参数 params:{"name":"天乔巴夏"}
     */
    private Object getParams(ProceedingJoinPoint joinPoint) {
        // 参数名
        String[] paramNames = getMethodSignature(joinPoint).getParameterNames();
        // 参数值
        Object[] paramValues = joinPoint.getArgs();
        // 存储参数
        Map<String, Object> params = new LinkedHashMap<>();
        for (int i = 0; i < paramNames.length; i++) {
            Object value = paramValues[i];
            // MultipartFile对象以文件名作为参数值
            if (value instanceof MultipartFile) {
                MultipartFile file = (MultipartFile) value;
                value = file.getOriginalFilename();
            }
            params.put(paramNames[i], value);
        }
        return params;
    }

    private MethodSignature getMethodSignature(ProceedingJoinPoint joinPoint) {
        return (MethodSignature) joinPoint.getSignature();
    }


    private HttpServletRequest getRequest() {
        ServletRequestAttributes requestAttributes =
                (ServletRequestAttributes) RequestContextHolder.getRequestAttributes();
        return requestAttributes.getRequest();
    }

}

bash复制代码POST http://localhost:8081/hello
Content-Type: application/json

{  "id" : 1,   "username" : "天乔巴夏",   "age": 18 }

json复制代码=================前置通知=====================
=================返回通知=====================
{"ipAddress":"127.0.0.1","description":"hello post","httpMethod":"POST","params":{"user":{"id":1,"age":18,"username":"天乔巴夏"}},"uri":"/hello","url":"http://localhost:8081/hello","result":"hello","methodType":"INSERT","startTime":1605596028383,"timeCost":28}
=================后置通知=====================

python复制代码def money_alloc(m, n):
    if n * 0.01 > m:
        raise ValueError("not enough money or too many people")
    result = []
    m = round(m, 2)
    while n > 1:
        # 这里需要注意两个细节：
        # - random.uniform(a, b)的随机区间是≥a&≤b,即[a, b]
        # - random.uniform(a, b)随机出来的值是0.0012032010230123，保留两位小数之后是有可能出现等于0.00的情况
        alloc_result = round(random.uniform(0.01, m-0.01), 2)
        # (m - alloc_result) < (n * 0.01)的判断是为了保证这一次的随机之后，后续的总金额可以继续分配，否则将重新随机指导满足这个条件
        if  (m - alloc_result) < (n * 0.01) or alloc_result <= 0.00:
            	continue
        result.append(alloc_result)
        n = n - 1
        m = m - alloc_result

    result.append(round(m, 2))
    return result

python复制代码for _ in xrange(10):
    print money_alloc(10, 5)

python复制代码[3.73, 6.15, 0.06, 0.03, 0.03]
[4.28, 0.8, 1.09, 2.13, 1.7]
[0.66, 2.27, 5.5, 1.5, 0.07]
[6.55, 1.46, 0.82, 0.2, 0.97]
[5.48, 0.47, 0.65, 0.48, 2.92]
[6.4, 3.09, 0.29, 0.01, 0.21]
[9.94, 0.02, 0.01, 0.01, 0.02]
[4.98, 4.97, 0.01, 0.01, 0.03]
[8.17, 1.3, 0.18, 0.17, 0.18]
[3.49, 5.45, 0.36, 0.3, 0.4]

python复制代码def money_alloc(m, n):
    if n * 0.01 > m:
        raise ValueError("not enough money or too many people")
    result = []
    m = round(m, 2)
    # 加入随机次数统计
    random_count = 0
    while n > 1:
        alloc_result = round(random.uniform(0.01, m-0.01), 2)
        if  (m - alloc_result) < (n * 0.01) or alloc_result <= 0.00:
            random_count += 1
            # 随机10次还没有结果，直接给丫来一个0.01，行不行?
            if random_count > 10:
                alloc_result = 0.01
                random_count = 0
                result.append(alloc_result)
                n = n - 1
                m = m - alloc_result
            continue
        result.append(alloc_result)
        n = n - 1
        m = m - alloc_result
    result.append(round(m, 2))
    return result

python复制代码for _ in xrange(10):
    print money_alloc(1, random.randint(10, 99))

python复制代码[0.03, 0.13, 0.16, 0.01, 0.1, 0.2, 0.06, 0.02, 0.01, 0.01, 0.01, 0.01, 0.02, 0.01, 0.02, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.03]

[0.79, 0.02, 0.03, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.03]

[0.01, 0.08, 0.01, 0.01, 0.02, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.01, 0.03]

python复制代码def money_alloc(m, n):
    if n * 0.01 > m:
        raise ValueError("not enough money or too many people")
    result = []
    m = round(m, 2)
    while n > 1:
        avg_money = round(m / n * 2, 2) - 0.01
        alloc_result = round(random.uniform(0.01, avg_money), 2)
        result.append(alloc_result)
        n = n - 1
        m = m - alloc_result
    result.append(round(m, 2))
    return result

bash复制代码# 10块钱5个人分
for _ in xrange(10):
    print money_alloc(10, 5)

# 分配结果, 随机结果在2附近的值一眼看下去还是居多的
[1.83, 0.78, 0.28, 2.74, 4.37]
[1.17, 4.13, 0.54, 0.66, 3.5]
[1.37, 1.67, 1.3, 5.57, 0.09]
[3.49, 2.5, 1.22, 0.75, 2.04]
[2.1, 3.2, 0.5, 3.19, 1.01]
[2.83, 2.01, 2.12, 1.2, 1.84]
[2.97, 0.79, 1.45, 1.52, 3.27]
[2.77, 1.64, 1.53, 0.41, 3.65]
[3.49, 0.88, 0.39, 3.26, 1.98]
[1.79, 3.61, 2.55, 1.21, 0.84]

bash复制代码# 1块钱50个人分，分配结果（数据太多，随机取了两个）
[0.01, 0.03, 0.01, 0.01, 0.01, 0.02, 0.03, 0.02, 0.03, 0.03, 0.01, 0.01, 0.01, 0.02, 0.02, 0.02, 0.01, 0.02, 0.02, 0.02, 0.02, 0.02, 0.03, 0.02, 0.02, 0.02, 0.02, 0.01, 0.02, 0.02, 0.03, 0.02, 0.02, 0.02, 0.02, 0.02, 0.03, 0.02, 0.01, 0.02, 0.03, 0.02, 0.01, 0.01, 0.02, 0.04, 0.03, 0.04, 0.01, 0.02]
[0.03, 0.03, 0.02, 0.02, 0.03, 0.03, 0.03, 0.02, 0.02, 0.01, 0.03, 0.02, 0.02, 0.02, 0.02, 0.01, 0.02, 0.01, 0.02, 0.03, 0.02, 0.01, 0.03, 0.02, 0.02, 0.03, 0.01, 0.02, 0.03, 0.02, 0.02, 0.01, 0.01, 0.03, 0.02, 0.01, 0.03, 0.02, 0.02, 0.02, 0.02, 0.01, 0.02, 0.02, 0.02, 0.01, 0.02, 0.01, 0.01, 0.02]

python复制代码def money_alloc(m, n):
    if n * 0.01 > m:
        raise ValueError("not enough money")
    # 0为线段的起点
    result = [0]
    m = round(m, 2)
    while n > 1:
        alloc_result = round(random.uniform(0.01, m - 0.01), 2)
        if alloc_result in result:
            continue
        result.append(alloc_result)
        n = n - 1
    # m为线段的终点
    result.append(m)
    result.sort()
    return [round(result[index+1]- item, 2) for index, item in enumerate(result) if index < len(result) - 1]

python复制代码[1.07, 6.08, 2.85]

[0.04, 0.11, 0.02, 0.02, 0.02, 0.01, 0.01, 0.01, 0.07, 0.02, 0.02, 0.05, 0.12, 0.02, 0.01, 0.01, 0.01, 0.13, 0.02, 0.01, 0.05, 0.03, 0.02, 0.07, 0.01, 0.02, 0.02, 0.01, 0.03, 0.01]

python复制代码for _ in xrange(5):
    a = time.time()
    money_alloc(10000, 30000)
    b = time.time()
    print b - a

python复制代码7.04587507248
7.84848403931
7.50485801697
7.98592209816
8.28649902344

python复制代码def money_alloc(m, n):
    if n * 0.01 > m:
        raise ValueError("not enough money")
    result = [0]
    # 牺牲一部分空间，提升查重的效率
    tmp_dict = dict()
    m = round(m, 2)
    while n > 1:
        alloc_result = round(random.uniform(0.01, m - 0.01), 2)
        # hash 版本
        if alloc_result in tmp_dict:
            continue
        tmp_dict[alloc_result] = None
        result.append(alloc_result)
        n = n - 1

    result.append(m)
    result.sort()
    return [round(result[index+1]- item, 2) for index, item in enumerate(result) if index < len(result) - 1]

python复制代码0.197105169296
0.169443130493
0.162744998932
0.167745113373
0.147526979446

python复制代码|  uniform(self, a, b)
     |      Get a random number in the range [a, b) or [a, b] depending on rounding.

python复制代码>>> random.uniform(0, 1)
0.15407896775722285
>>> random.uniform(0, 1)
0.16189270320003113
>>> random.uniform(0, 0)    # a == b
0.0
>>> random.uniform(0, -1)    # a > b
-0.8838459569306347
>>> random.uniform(0, -100)     # a > b
-76.93918157758513

java复制代码@Component
public class SignAuthFilter extends OncePerRequestFilter{
static final String FAVICON = "/favicon.ico";
static final String PREFIX = "attack:signature:";
}

xml复制代码<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.1.7</version>
    <relativePath/> <!-- lookup parent from update -->
</parent>

<properties>
    <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
    <spring-cloud.version>Finchley.RELEASE</spring-cloud.version>
    <java.version>1.8</java.version>
</properties>

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
</dependencies>

<build>
    <pluginManagement>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </pluginManagement>
</build>
<repositories>
    <repository>
        <id>spring-milestones</id>
        <name>Spring Milestones</name>
        <url>https://repo.spring.io/milestone</url>
        <snapshots>
            <enabled>false</enabled>
        </snapshots>
    </repository>
</repositories>

less复制代码/**
 * Iterate over registered {@link HandlerMethodArgumentResolver}s and invoke the one that supports it.
 * @throws IllegalStateException if no suitable {@link HandlerMethodArgumentResolver} is found.
 */
@Override
@Nullable
public Object resolveArgument(MethodParameter parameter, @Nullable ModelAndViewContainer mavContainer,
        NativeWebRequest webRequest, @Nullable WebDataBinderFactory binderFactory) throws Exception {

    HandlerMethodArgumentResolver resolver = getArgumentResolver(parameter);
    if (resolver == null) {
        throw new IllegalArgumentException("Unknown parameter type [" + parameter.getParameterType().getName() + "]");
    }
    return resolver.resolveArgument(parameter, mavContainer, webRequest, binderFactory);
}

less复制代码@Target(ElementType.PARAMETER)
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface ListParam {
    /**
     * Alias for {@link #name}.
     */
    @AliasFor("name") String value() default "";

    /**
     * The name of the request parameter to bind to.
     *
     * @since 4.2
     */
    @AliasFor("value") String name() default "";
}

ini复制代码public class ListHandlerMethodArgumentResolver implements HandlerMethodArgumentResolver {
    @Override
    public boolean supportsParameter(MethodParameter parameter) {
        return parameter.hasParameterAnnotation(ListParam.class);
    }

    @Override
    public Object resolveArgument(MethodParameter parameter, ModelAndViewContainer mavContainer,
            NativeWebRequest webRequest, WebDataBinderFactory binderFactory) throws Exception {
        ListParam param = parameter.getParameterAnnotation(ListParam.class);
        if (param == null) {
            throw new IllegalArgumentException(
                    "Unknown parameter type [" + parameter.getParameterType().getName() + "]");
        }

        String name = "".equalsIgnoreCase(param.name()) ? param.value() : param.name();
        if ("".equalsIgnoreCase(name)) {
            name = parameter.getParameter().getName();
        }
        String ans = webRequest.getParameter(name);
        if (ans == null) {
            return null;
        }

        String[] cells = StringUtils.split(ans, ",");
        return Arrays.asList(cells);
    }
}

scala复制代码@SpringBootApplication
public class Application extends WebMvcConfigurationSupport {

    @Override
    protected void addArgumentResolvers(List<HandlerMethodArgumentResolver> argumentResolvers) {
        argumentResolvers.add(new ListHandlerMethodArgumentResolver());
    }

    public static void main(String[] args) {
        SpringApplication.run(Application.class);
    }
}

typescript复制代码@RestController
@RequestMapping(path = "get")
public class ParamGetRest {
    /**
     * 自定义参数解析器
     *
     * @param names
     * @param age
     * @return
     */
    @GetMapping(path = "self")
    public String selfParam(@ListParam(name = "names") List<String> names, Integer age) {
        return names + " | age=" + age;
    }

    @GetMapping(path = "self2")
    public String selfParam2(List<String> names, Integer age) {
        return names + " | age=" + age;
    }
}

xml复制代码switch-master <master name> <oldip> <oldport> <newip> <newport>

bash复制代码vim /usr/lib/systemd/system/docker.service
# 修改ExecStart这行，ExecStart= xxxxx不要改变直接空格之后添加
# tcp是开启远程访问，unix是本机访问，不开启本机访问，在服务器上使用时会有以下错误
# Cannot connect to the Docker daemon at unix:///var/run/docker.sock. Is the docker daemon running?
ExecStart=/usr/bin/dockerd  -H tcp://0.0.0.0:2375  -H unix:///var/run/docker.sock 

#重新加载配置文件                  
systemctl daemon-reload 
# 重启服务 
systemctl restart docker.service 
# 查看端口是否开启 默认端口2375 
netstat -nptl
#直接curl看是否生效 如果没有生效显示被拒绝，生效之后显示json
curl http://127.0.0.1:2375/info

java复制代码import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author smallthanks
 */
@SpringBootApplication
@RestController
public class DemoApplication {

	public static void main(String[] args) {
		SpringApplication.run(DemoApplication.class, args);
	}

	@GetMapping("/index")
	public String index() {
	    return "hello world";
    }
}

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 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.4.2</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>

    <packaging>jar</packaging>
    <groupId>com.example</groupId>
    <artifactId>demo</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>demo</name>
    <description>Demo project for Spring Boot</description>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
        <java.version>1.8</java.version>
    </properties>
    <repositories>
        <!--        使用阿里云的镜像-->
        <repository>
            <id>spring</id>
            <url>https://maven.aliyun.com/repository/spring</url>
            <releases>
                <enabled>true</enabled>
            </releases>
            <snapshots>
                <enabled>true</enabled>
            </snapshots>
        </repository>
    </repositories>
    <dependencies>
        ·····<!-此处为了读者简洁，省去了必要的依赖-->
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>
</project>

dockerfile复制代码# DockerFile的文件内容为，其字段的名称及作用自行Google，并存放于上图的目录
#指定基础镜像，在其上进行定制
FROM java:8
#维护者信息
MAINTAINER SmallThanks <SmallThanks@163.com>
#这里的 /tmp 目录就会在运行时自动挂载为匿名卷，任何向 /data 中写入的信息都不会记录进容器存储层
VOLUME /tmp
#复制上下文目录下的target/demo-1.0.0.jar 到容器里
COPY /target/demo-0.0.1-SNAPSHOT.jar demo-0.0.1-SNAPSHOT.jar
#bash方式执行，使demo-1.0.0.jar可访问
#RUN新建立一层，在其上执行这些命令，执行结束后， commit 这一层的修改，构成新的镜像。
RUN bash -c "touch /demo-0.0.1-SNAPSHOT.jar"

#声明运行时容器提供服务端口，这只是一个声明，在运行时并不会因为这个声明应用就会开启这个端口的服务
EXPOSE 8080

#指定容器启动程序及参数   <ENTRYPOINT> "<CMD>"
ENTRYPOINT ["java","-jar","demo-0.0.1-SNAPSHOT.jar"]

复制代码我们也可以通过其他的手段实现自动部署，但我我还没学会。

ini复制代码etc/my.cnf

server-id=1
log-bin=mysql-bin
binlog-format=ROW

ini复制代码 ## mysql serverId
 canal.instance.mysql.slaveId = 1234
 #position info，需要修改成自己的数据库信息
 canal.instance.master.address = 127.0.0.1:3306
 canal.instance.master.journal.name =
 canal.instance.master.position =
 canal.instance.master.timestamp =
 #canal.instance.standby.address =
 #canal.instance.standby.journal.name =
 #canal.instance.standby.position =
 #canal.instance.standby.timestamp =
 #username/password，需要修改成自己的数据库信息
 canal.instance.dbUsername = canal
 canal.instance.dbPassword = canal
 canal.instance.defaultDatabaseName =
 canal.instance.connectionCharset = UTF-8
 #table regex
 canal.instance.filter.regex = .\*\\\\..\*

bash复制代码 sh bin/startup.sh
 关闭使用 bin/stop.sh

xml复制代码 <dependency>
   <groupId>com.alibaba.otter</groupId>
   <artifactId>canal.client</artifactId>
   <version>1.0.12</version>
 </dependency>

ini复制代码public class Demo {
 public static void main(String[] args) {
     //创建连接
     CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("hadoop03", 11111),
             "example", "", "");
     connector.connect();
     //订阅
     connector.subscribe();
     connector.rollback();
     int batchSize = 1000;
     int emptyCount = 0;
     int totalEmptyCount = 100;
     while (totalEmptyCount > emptyCount) {
         Message msg = connector.getWithoutAck(batchSize);
         long id = msg.getId();
         List<CanalEntry.Entry> entries = msg.getEntries();
         if(id == -1 || entries.size() == 0){
             emptyCount++;
             System.out.println("emptyCount : " + emptyCount);
             try {
                 Thread.sleep(3000);
             } catch (InterruptedException e) {
                 e.printStackTrace();
             }
         }else{
             emptyCount = 0;
             printEntry(entries);
         }
         connector.ack(id);
     }
 }
 // batch -> entries -> rowchange - rowdata -> cols
 private static void printEntry(List<CanalEntry.Entry> entries) {
     for (CanalEntry.Entry entry : entries){
         if(entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN ||
                 entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND){
             continue;
         }
         CanalEntry.RowChange rowChange = null;
         try {
             rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
         } catch (InvalidProtocolBufferException e) {
             e.printStackTrace();
         }
         CanalEntry.EventType eventType = rowChange.getEventType();
         System.out.println(entry.getHeader().getLogfileName()+" __ " +
                 entry.getHeader().getSchemaName() + " __ " + eventType);
         List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
         for(CanalEntry.RowData rowData : rowDatasList){
             for(CanalEntry.Column column: rowData.getAfterColumnsList()){
                 System.out.println(column.getName() + " - " +
                         column.getValue() + " - " +
                         column.getUpdated());
             }
         }
     }
 }
}

ini复制代码 canal.zkServers = hadoop02:2181,hadoop03:2181,hadoop04:2181
 canal.serverMode = kafka
 canal.mq.servers = hadoop02:9092,hadoop03:9092,hadoop04:9092

ini复制代码  canal.instance.mysql.slaveId = 790 /两台canal server的slaveID唯一
  canal.mq.topic = canal_log //指定将数据发送到kafka的topic

csharp复制代码// Added Dependencies
implementation "androidx.recyclerview:recyclerview:1.1.0"
implementation 'android.arch.lifecycle:extensions:1.1.1'
implementation 'androidx.lifecycle:lifecycle-viewmodel-ktx:2.2.0'
implementation 'androidx.lifecycle:lifecycle-runtime-ktx:2.2.0'
implementation 'com.github.bumptech.glide:glide:4.11.0'

//retrofit
implementation 'com.squareup.retrofit2:retrofit:2.8.1'
implementation "com.squareup.retrofit2:converter-moshi:2.6.2"

//Coroutine
implementation "org.jetbrains.kotlinx:kotlinx-coroutines-android:1.3.6"
implementation "org.jetbrains.kotlinx:kotlinx-coroutines-core:1.3.6"

kotlin复制代码https://reqres.in/api/users

kotlin复制代码package com.my.mvi.data.model

data class User(
    @Json(name = "id")
    val id: Int = 0,
    @Json(name = "first_name")
    val name: String = "",
    @Json(name = "email")
    val email: String = "",
    @Json(name = "avator")
    val avator: String = ""
)

kotlin复制代码package com.my.mvi.data.api

interface ApiService {

   @GET("users")
   suspend fun getUsers(): List<User>
}

kotlin复制代码
object RetrofitBuilder {

    private const val BASE_URL = "https://reqres.in/api/user/1"

    private fun getRetrofit() = Retrofit.Builder()
        .baseUrl(BASE_URL)
        .addConverterFactory(MoshiConverterFactory.create())
        .build()


    val apiService: ApiService = getRetrofit().create(ApiService::class.java)

}

kotlin复制代码package com.my.mvi.data.repository

class MainRepository(private val apiService: ApiService) {

    suspend fun getUsers() = apiService.getUsers()

}

kotlin复制代码package com.my.mvi.ui.main.adapter

class MainAdapter(
    private val users: ArrayList<User>
) : RecyclerView.Adapter<MainAdapter.DataViewHolder>() {

    class DataViewHolder(itemView: View) : RecyclerView.ViewHolder(itemView) {
        fun bind(user: User) {
            itemView.textViewUserName.text = user.name
            itemView.textViewUserEmail.text = user.email
            Glide.with(itemView.imageViewAvatar.context)
                .load(user.avatar)
                .into(itemView.imageViewAvatar)
        }
    }

    override fun onCreateViewHolder(parent: ViewGroup, viewType: Int) =
        DataViewHolder(
            LayoutInflater.from(parent.context).inflate(
                R.layout.item_layout, parent,
                false
            )
        )

    override fun getItemCount(): Int = users.size

    override fun onBindViewHolder(holder: DataViewHolder, position: Int) =
        holder.bind(users[position])

    fun addData(list: List<User>) {
        users.addAll(list)
    }

}

xml复制代码<?xml version="1.0" encoding="utf-8"?>
<androidx.constraintlayout.widget.ConstraintLayout xmlns:android="http://schemas.android.com/apk/res/android"
    xmlns:app="http://schemas.android.com/apk/res-auto"
    xmlns:tools="http://schemas.android.com/tools"
    android:id="@+id/container"
    android:layout_width="match_parent"
    android:layout_height="60dp">

    <ImageView
        android:id="@+id/imageViewAvatar"
        android:layout_width="60dp"
        android:layout_height="0dp"
        android:padding="4dp"
        app:layout_constraintBottom_toBottomOf="parent"
        app:layout_constraintStart_toStartOf="parent"
        app:layout_constraintTop_toTopOf="parent" />

    <androidx.appcompat.widget.AppCompatTextView
        android:id="@+id/textViewUserName"
        style="@style/TextAppearance.AppCompat.Large"
        android:layout_width="0dp"
        android:layout_height="wrap_content"
        android:layout_marginStart="8dp"
        android:layout_marginTop="4dp"
        app:layout_constraintEnd_toEndOf="parent"
        app:layout_constraintStart_toEndOf="@+id/imageViewAvatar"
        app:layout_constraintTop_toTopOf="parent"/>

    <androidx.appcompat.widget.AppCompatTextView
        android:id="@+id/textViewUserEmail"
        android:layout_width="0dp"
        android:layout_height="wrap_content"
        app:layout_constraintEnd_toEndOf="parent"
        app:layout_constraintStart_toStartOf="@+id/textViewUserName"
        app:layout_constraintTop_toBottomOf="@+id/textViewUserName" />

</androidx.constraintlayout.widget.ConstraintLayout>

kotlin复制代码package com.my.mvi.ui.main.intent

sealed class MainIntent {

    object FetchUser : MainIntent()

}

kotlin复制代码sealed class MainState {

    object Idle : MainState()
    object Loading : MainState()
    data class Users(val user: List<User>) : MainState()
    data class Error(val error: String?) : MainState()

}

kotlin复制代码package com.my.mvi.ui.main.viewmodel

class MainViewModel(
    private val repository: MainRepository
) : ViewModel() {

    val userIntent = Channel<MainIntent>(Channel.UNLIMITED)
    private val _state = MutableStateFlow<MainState>(MainState.Idle)
    val state: StateFlow<MainState>
        get() = _state

    init {
        handleIntent()
    }

    private fun handleIntent() {
        viewModelScope.launch {
            userIntent.consumeAsFlow().collect {
                when (it) {
                    is MainIntent.FetchUser -> fetchUser()
                }
            }
        }
    }

    private fun fetchUser() {
        viewModelScope.launch {
            _state.value = MainState.Loading
            _state.value = try {
                MainState.Users(repository.getUsers())
            } catch (e: Exception) {
                MainState.Error(e.localizedMessage)
            }
        }
    }
}

kotlin复制代码class ViewModelFactory(private val apiService: ApiService) : ViewModelProvider.Factory {

    override fun <T : ViewModel?> create(modelClass: Class<T>): T {
        if (modelClass.isAssignableFrom(MainViewModel::class.java)) {
            return MainViewModel(MainRepository(apiService)) as T
        }
        throw IllegalArgumentException("Unknown class name")
    }

}

kotlin复制代码package com.my.mvi.ui.main.view

class MainActivity : AppCompatActivity() {

    private lateinit var mainViewModel: MainViewModel
    private var adapter = MainAdapter(arrayListOf())

    override fun onCreate(savedInstanceState: Bundle?) {
        super.onCreate(savedInstanceState)
        setContentView(R.layout.activity_main)
        setupUI()
        setupViewModel()
        observeViewModel()
        setupClicks()
    }

    private fun setupClicks() {
        buttonFetchUser.setOnClickListener {
            lifecycleScope.launch {
                mainViewModel.userIntent.send(MainIntent.FetchUser)
            }
        }
    }


    private fun setupUI() {
        recyclerView.layoutManager = LinearLayoutManager(this)
        recyclerView.run {
            addItemDecoration(
                DividerItemDecoration(
                    recyclerView.context,
                    (recyclerView.layoutManager as LinearLayoutManager).orientation
                )
            )
        }
        recyclerView.adapter = adapter
    }


    private fun setupViewModel() {
        mainViewModel = ViewModelProviders.of(
            this,
            ViewModelFactory(
                ApiHelperImpl(
                    RetrofitBuilder.apiService
                )
            )
        ).get(MainViewModel::class.java)
    }

    private fun observeViewModel() {
        lifecycleScope.launch {
            mainViewModel.state.collect {
                when (it) {
                    is MainState.Idle -> {

                    }
                    is MainState.Loading -> {
                        buttonFetchUser.visibility = View.GONE
                        progressBar.visibility = View.VISIBLE
                    }

                    is MainState.Users -> {
                        progressBar.visibility = View.GONE
                        buttonFetchUser.visibility = View.GONE
                        renderList(it.user)
                    }
                    is MainState.Error -> {
                        progressBar.visibility = View.GONE
                        buttonFetchUser.visibility = View.VISIBLE
                        Toast.makeText(this@MainActivity, it.error, Toast.LENGTH_LONG).show()
                    }
                }
            }
        }
    }

    private fun renderList(users: List<User>) {
        recyclerView.visibility = View.VISIBLE
        users.let { listOfUsers -> listOfUsers.let { adapter.addData(it) } }
        adapter.notifyDataSetChanged()
    }
}

xml复制代码<?xml version="1.0" encoding="utf-8"?>
<androidx.constraintlayout.widget.ConstraintLayout xmlns:android="http://schemas.android.com/apk/res/android"
    xmlns:app="http://schemas.android.com/apk/res-auto"
    xmlns:tools="http://schemas.android.com/tools"
    android:layout_width="match_parent"
    android:layout_height="match_parent"
    tools:context=".ui.main.view.MainActivity">

    <androidx.recyclerview.widget.RecyclerView
        android:id="@+id/recyclerView"
        android:layout_width="match_parent"
        android:layout_height="match_parent"
        android:visibility="gone" />

    <ProgressBar
        android:id="@+id/progressBar"
        android:layout_width="wrap_content"
        android:layout_height="wrap_content"
        app:layout_constraintBottom_toBottomOf="parent"
        app:layout_constraintEnd_toEndOf="parent"
        android:visibility="gone"
        app:layout_constraintStart_toStartOf="parent"
        app:layout_constraintTop_toTopOf="parent" />

    <Button
        android:id="@+id/buttonFetchUser"
        android:layout_width="wrap_content"
        android:layout_height="wrap_content"
        android:text="@string/fetch_user"
        app:layout_constraintBottom_toBottomOf="parent"
        app:layout_constraintEnd_toEndOf="parent"
        app:layout_constraintStart_toStartOf="parent"
        app:layout_constraintTop_toTopOf="parent" />

</androidx.constraintlayout.widget.ConstraintLayout>

java复制代码public interface ConsumerStrategy<P, C> {

    /**
     * 暴露当前策略的钩子
     *
     * @return 判断钩子
     */
    Predicate<P> predicate();

    /**
     * 暴露当前策略的消费逻辑
     *
     * @return 消费逻辑
     */
    Consumer<C> consumer();

    /**
     * 真实的消费数据
     *
     * @param p 钩子来源
     * @param c 消费来源
     */
    default void handle(P p, C c) {
        if (this.predicate().test(p)) {
            this.consumer().accept(c);
        }
    }
}

java复制代码/**
 * 简化处理
 * String 代表来自区域风险系数
 * Integer 打工人
 */
enum ReturnHomeStrategy implements ConsumerStrategy<String, Integer> {
    /**
     * 高风险
     */
    HIGH_RISK(from -> "HIGH_RISK".equals(from), i -> {
        throw new RuntimeException("滚!");
    }),
    /**
     * 中风险
     */
    MIDDLE_RISK(from -> "MIDDLE_RISK".equals(from), i -> {
        throw new RuntimeException("滚!");
    }),
    /**
     * 低风险
     */
    LOW_RISK(from -> "LOW_RISK".equals(from), i -> {
        //todo 如果可以,也请滚
        //todo 核算检测
        //todo 居家隔离
    });

    private final Predicate<String> predicate;

    private final Consumer<Integer> consumer;

    ReturnHomeStrategy(Predicate<String> predicate, Consumer<Integer> consumer) {
        this.predicate = predicate;
        this.consumer = consumer;
    }

    @Override
    public Predicate<String> predicate() {
        return this.predicate;
    }

    @Override
    public Consumer<Integer> consumer() {
        return this.consumer;
    }
}

java复制代码/**
 * 某人回家
 *
 * @param from 代表来自区域风险系数
 * @param id   打工人
 */
public void returnHome(String from, Integer id) {
    for (ReturnHomeStrategy value : ReturnHomeStrategy.values()) {
        value.handle(from, id);
    }
}

java复制代码public interface FunctionStrategy<P, T, R> {

    /**
     * 暴露当前策略的钩子
     *
     * @return 判断钩子
     */
    Predicate<P> predicate();

    /**
     * 暴露当前策略的生产逻辑
     *
     * @return 消费逻辑
     */
    Function<T, R> function();
}

java复制代码/**
 * 简化处理
 * String 代表来自区域风险系数
 * Integer 打工人
 * String 村里开具证明
 */
enum ReturnHomeStrategy implements FunctionStrategy<String, Integer, String> {
    /**
     * 高风险
     */
    HIGH_RISK(from -> "HIGH_RISK".equals(from), i -> "滚!"),
    /**
     * 中风险
     */
    MIDDLE_RISK(from -> "MIDDLE_RISK".equals(from), i -> "滚!"),
    /**
     * 低风险
     */
    LOW_RISK(from -> "LOW_RISK".equals(from), i -> "村里的祸害回来了,给他做个核算检测!");

    private final Predicate<String> predicate;

    private final Function<Integer, String> function;

    public Predicate<String> getPredicate() {
        return predicate;
    }

    public Function<Integer, String> getFunction() {
        return function;
    }

    ReturnHomeStrategy(Predicate<String> predicate, Function<Integer, String> function) {
        this.predicate = predicate;
        this.function = function;
    }

    @Override
    public Predicate<String> predicate() {
        return this.predicate;
    }

    @Override
    public Function<Integer, String> function() {
        return this.function;
    }
}

java复制代码/**
 * 某人回家
 *
 * @param from 代表来自区域风险系数
 * @param id   打工人
 * @return 要证明
 */
public String returnHome(String from, Integer id) {
    for (ReturnHomeStrategy value : ReturnHomeStrategy.values()) {
        if (value.predicate().test(from)) {
            return value.function().apply(id);
        }
    }
    throw new RuntimeException("外星人,抓起来放进动物园卖门票!");
}