javascript复制代码// gRPC服务，这是个在线商城的库存服务
service StockService {
    // 双向流式：批量扣减库存
    rpc BatchDeduct (stream ProductOrder) returns (stream DeductReply) {}
}

// 扣减库存返回结果的数据结构
message DeductReply {
    // 返回码
    int32 code = 1;
    // 描述信息
    string message = 2;
}

groovy复制代码// 使用springboot插件
plugins {
    id 'org.springframework.boot'
}

dependencies {
    implementation 'org.projectlombok:lombok'
    implementation 'org.springframework.boot:spring-boot-starter'
    // 作为gRPC服务提供方，需要用到此库
    implementation 'net.devh:grpc-server-spring-boot-starter'
    // 依赖自动生成源码的工程
    implementation project(':grpc-lib')
    // annotationProcessor不会传递，使用了lombok生成代码的模块，需要自己声明annotationProcessor
    annotationProcessor 'org.projectlombok:lombok'
}

yml复制代码spring:
  application:
    name: double-stream-server-side
# gRPC有关的配置，这里只需要配置服务端口号
grpc:
  server:
    port: 9901

java复制代码package grpctutorials;

import com.bolingcavalry.grpctutorials.lib.DeductReply;
import com.bolingcavalry.grpctutorials.lib.ProductOrder;
import com.bolingcavalry.grpctutorials.lib.StockServiceGrpc;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;
import net.devh.boot.grpc.server.service.GrpcService;

@GrpcService
@Slf4j
public class GrpcServerService extends StockServiceGrpc.StockServiceImplBase {

    @Override
    public StreamObserver<ProductOrder> batchDeduct(StreamObserver<DeductReply> responseObserver) {
        // 返回匿名类，给上层框架使用
        return new StreamObserver<ProductOrder>() {

            private int totalCount = 0;

            @Override
            public void onNext(ProductOrder value) {
                log.info("正在处理商品[{}]，数量为[{}]",
                        value.getProductId(),
                        value.getNumber());

                // 增加总量
                totalCount += value.getNumber();

                int code;
                String message;

                // 假设单数的都有库存不足的问题
                if (0 == value.getNumber() % 2) {
                    code = 10000;
                    message = String.format("商品[%d]扣减库存数[%d]成功", value.getProductId(), value.getNumber());
                } else {
                    code = 10001;
                    message = String.format("商品[%d]扣减库存数[%d]失败", value.getProductId(), value.getNumber());
                }

                responseObserver.onNext(DeductReply.newBuilder()
                        .setCode(code)
                        .setMessage(message)
                        .build());
            }

            @Override
            public void onError(Throwable t) {
                log.error("批量减扣库存异常", t);
            }

            @Override
            public void onCompleted() {
                log.info("批量减扣库存完成，共计[{}]件商品", totalCount);
                responseObserver.onCompleted();
            }
        };
    }
}

groovy复制代码plugins {
    id 'org.springframework.boot'
}

dependencies {
    implementation 'org.projectlombok:lombok'
    implementation 'org.springframework.boot:spring-boot-starter'
    implementation 'org.springframework.boot:spring-boot-starter-web'
    implementation 'net.devh:grpc-client-spring-boot-starter'
    implementation project(':grpc-lib')
}

yml复制代码server:
  port: 8082
spring:
  application:
    name: double-stream-client-side

grpc:
  client:
    # gRPC配置的名字，GrpcClient注解会用到
    double-stream-server-side:
      # gRPC服务端地址
      address: 'static://127.0.0.1:9901'
      enableKeepAlive: true
      keepAliveWithoutCalls: true
      negotiationType: plaintext

java复制代码package com.bolingcavalry.grpctutorials;

import io.grpc.stub.StreamObserver;

public interface ExtendResponseObserver<T> extends StreamObserver<T> {
    String getExtra();
}

java复制代码package grpctutorials;

import com.bolingcavalry.grpctutorials.lib.DeductReply;
import com.bolingcavalry.grpctutorials.lib.ProductOrder;
import com.bolingcavalry.grpctutorials.lib.StockServiceGrpc;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;
import net.devh.boot.grpc.client.inject.GrpcClient;
import org.springframework.stereotype.Service;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

@Service
@Slf4j
public class GrpcClientService {

    @GrpcClient("double-stream-server-side")
    private StockServiceGrpc.StockServiceStub stockServiceStub;

    /**
     * 批量减库存
     * @param count
     * @return
     */
    public String batchDeduct(int count) {

        CountDownLatch countDownLatch = new CountDownLatch(1);

        // responseObserver的onNext和onCompleted会在另一个线程中被执行，
        // ExtendResponseObserver继承自StreamObserver
        ExtendResponseObserver<DeductReply> responseObserver = new ExtendResponseObserver<DeductReply>() {

            // 用stringBuilder保存所有来自服务端的响应
            private StringBuilder stringBuilder = new StringBuilder();

            @Override
            public String getExtra() {
                return stringBuilder.toString();
            }

            /**
             * 客户端的流式请求期间，每一笔请求都会收到服务端的一个响应，
             * 对应每个响应，这里的onNext方法都会被执行一次，入参是响应内容
             * @param value
             */
            @Override
            public void onNext(DeductReply value) {
                log.info("batch deduct on next");
                // 放入匿名类的成员变量中
                stringBuilder.append(String.format("返回码[%d]，返回信息:%s<br>" , value.getCode(), value.getMessage()));
            }

            @Override
            public void onError(Throwable t) {
                log.error("batch deduct gRPC request error", t);
                stringBuilder.append("batch deduct gRPC error, " + t.getMessage());
                countDownLatch.countDown();
            }

            /**
             * 服务端确认响应完成后，这里的onCompleted方法会被调用
             */
            @Override
            public void onCompleted() {
                log.info("batch deduct on complete");
                // 执行了countDown方法后，前面执行countDownLatch.await方法的线程就不再wait了，
                // 会继续往下执行
                countDownLatch.countDown();
            }
        };

        // 远程调用，此时数据还没有给到服务端
        StreamObserver<ProductOrder> requestObserver = stockServiceStub.batchDeduct(responseObserver);

        for(int i=0; i<count; i++) {
            // 每次执行onNext都会发送一笔数据到服务端，
            // 服务端的onNext方法都会被执行一次
            requestObserver.onNext(build(101 + i, 1 + i));
        }

        // 客户端告诉服务端：数据已经发完了
        requestObserver.onCompleted();

        try {
            // 开始等待，如果服务端处理完成，那么responseObserver的onCompleted方法会在另一个线程被执行，
            // 那里会执行countDownLatch的countDown方法，一但countDown被执行，下面的await就执行完毕了，
            // await的超时时间设置为2秒
            countDownLatch.await(2, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            log.error("countDownLatch await error", e);
        }

        log.info("service finish");
        // 服务端返回的内容被放置在requestObserver中，从getExtra方法可以取得
        return responseObserver.getExtra();
    }

    /**
     * 创建ProductOrder对象
     * @param productId
     * @param num
     * @return
     */
    private static ProductOrder build(int productId, int num) {
        return ProductOrder.newBuilder().setProductId(productId).setNumber(num).build();
    }
}

java复制代码package grpctutorials;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class GrpcClientController {

    @Autowired
    private GrpcClientService grpcClientService;

    @RequestMapping("/")
    public String printMessage(@RequestParam(defaultValue = "1") int count) {
        return grpcClientService.batchDeduct(count);
    }
}

java复制代码// 
- After=2017-01-20T17:42:47.789-07:00[America/Denver]

// 
- Before=2017-01-20T17:42:47.789-07:00[America/Denver]

//
- Between=2017-01-20T17:42:47.789-07:00[America/Denver], 2017-01-21T17:42:47.789-07:00[America/Denver]

// 
- Cookie=chocolate, ch.p

java复制代码protected Mono<?> getHandlerInternal(ServerWebExchange exchange) {
   // don't handle requests on management port if set and different than server port
   if (this.managementPortType == DIFFERENT && this.managementPort != null
         && exchange.getRequest().getURI().getPort() == this.managementPort) {
      return Mono.empty();
   }
   exchange.getAttributes().put(GATEWAY_HANDLER_MAPPER_ATTR, getSimpleName());

   return lookupRoute(exchange)
         // .log("route-predicate-handler-mapping", Level.FINER) //name this
         .flatMap((Function<Route, Mono<?>>) r -> {
            exchange.getAttributes().remove(GATEWAY_PREDICATE_ROUTE_ATTR);
            if (logger.isDebugEnabled()) {
               logger.debug("Mapping [" + getExchangeDesc(exchange) + "] to " + r);
            }

            exchange.getAttributes().put(GATEWAY_ROUTE_ATTR, r);
            return Mono.just(webHandler);
         }).switchIfEmpty(Mono.empty().then(Mono.fromRunnable(() -> {
            exchange.getAttributes().remove(GATEWAY_PREDICATE_ROUTE_ATTR);
            if (logger.isTraceEnabled()) {
               logger.trace("No RouteDefinition found for [" + getExchangeDesc(exchange) + "]");
            }
         })));
}

java复制代码protected Mono<Route> lookupRoute(ServerWebExchange exchange) {
   return this.routeLocator.getRoutes()
         // individually filter routes so that filterWhen error delaying is not a
         // problem
         .concatMap(route -> Mono.just(route).filterWhen(r -> {
            // add the current route we are testing
            exchange.getAttributes().put(GATEWAY_PREDICATE_ROUTE_ATTR, r.getId());
            return r.getPredicate().apply(exchange);
         })
               // instead of immediately stopping main flux due to error, log and
               // swallow it
               .doOnError(e -> logger.error("Error applying predicate for route: " + route.getId(), e))
               .onErrorResume(e -> Mono.empty()))
         // .defaultIfEmpty() put a static Route not found
         // or .switchIfEmpty()
         // .switchIfEmpty(Mono.<Route>empty().log("noroute"))
         .next()
         // TODO: error handling
         .map(route -> {
            validateRoute(route, exchange);
            return route;
         });


}

java复制代码private static class DefaultGatewayFilterChain implements GatewayFilterChain {
    
   // 当前 Filter 链索引 
   private final int index;
   // Filter 集合 
   private final List<GatewayFilter> filters;

   DefaultGatewayFilterChain(List<GatewayFilter> filters) {
      this.filters = filters;
      this.index = 0;
   }

   private DefaultGatewayFilterChain(DefaultGatewayFilterChain parent, int index) {
      this.filters = parent.getFilters();
      this.index = index;
   }

   public List<GatewayFilter> getFilters() {
      return filters;
   }

   @Override
   public Mono<Void> filter(ServerWebExchange exchange) {
      return Mono.defer(() -> {
         if (this.index < filters.size()) {
            // 逐个 Filter 过滤调用 
            GatewayFilter filter = filters.get(this.index);
            DefaultGatewayFilterChain chain = new DefaultGatewayFilterChain(this,
                  this.index + 1);
            return filter.filter(exchange, chain);
         }
         else {
            return Mono.empty(); // complete
         }
      });
   }

}

java复制代码public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {

    // 通常判断部分条件  , 如果该 Filter 不符合 , 则跳过该 Filter
   if (isAlreadyRouted(exchange)
         || (!"http".equals(scheme) && !"https".equals(scheme))) {
      return chain.filter(exchange);
   }

java复制代码C- NettyRoutingFilter
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
   // 请求 URL : http://httpbin.org:80/get
   URI requestUrl = exchange.getRequiredAttribute(GATEWAY_REQUEST_URL_ATTR);
    // 协议类型 : http
   String scheme = requestUrl.getScheme();
   
   // Step 1 : filter 链处理 ,如果不符合 http 协议 , 就通过下一个 Filter 处理
   if (isAlreadyRouted(exchange)
         || (!"http".equals(scheme) && !"https".equals(scheme))) {
      return chain.filter(exchange);
   }
   // Step 2 : 标识 Routed 已处理
   setAlreadyRouted(exchange);

   // Step 3 : 获取 Request 请求对象 , 这个是外部请求的对象
   ServerHttpRequest request = exchange.getRequest();
    
   // Step 4 : 获取 Method 类型 (get/post...)
   final HttpMethod method = HttpMethod.valueOf(request.getMethodValue());
   final String url = requestUrl.toString();
   
   // Step 5 : 对 Header 进行处理 , 需要转发过去
   HttpHeaders filtered = filterRequest(getHeadersFilters(), exchange);
   final DefaultHttpHeaders httpHeaders = new DefaultHttpHeaders();
   filtered.forEach(httpHeaders::set);
    
   // -> Transfer-Encoding
   String transferEncoding = request.getHeaders().getFirst(HttpHeaders.TRANSFER_ENCODING);
   boolean chunkedTransfer = "chunked".equalsIgnoreCase(transferEncoding);
   // -> preserveHostHeader
   boolean preserveHost = exchange.getAttributeOrDefault(PRESERVE_HOST_HEADER_ATTRIBUTE, false);

   // 通过 netty httpClient 发起转发请求 , PS !!! 这里是异步的
   Flux<HttpClientResponse> responseFlux = this.httpClient
         .chunkedTransfer(chunkedTransfer).request(method).uri(url)
         .send((req, nettyOutbound) -> {
            // Step 6 : 转发 Header 
            req.headers(httpHeaders);
            
            // => 是否需要记录之前的 host
            if (preserveHost) {
               String host = request.getHeaders().getFirst(HttpHeaders.HOST);
               req.header(HttpHeaders.HOST, host);
            }
            
            // Step 7 : 真正发起请求
            return nettyOutbound.options(NettyPipeline.SendOptions::flushOnEach)
                  .send(request.getBody()
                        .map(dataBuffer -> ((NettyDataBuffer) dataBuffer)
                              .getNativeBuffer()));
         }).responseConnection((res, connection) -> {
            // Step 8 : 请求完成 , 获取 response
            ServerHttpResponse response = exchange.getResponse();
            
            // Step 9 : 转发headers 和 status 等属性
            HttpHeaders headers = new HttpHeaders();
            res.responseHeaders().forEach(
                  entry -> headers.add(entry.getKey(), entry.getValue()));
            
            // => String CONTENT_TYPE = "Content-Type" 
            // => String ORIGINAL_RESPONSE_CONTENT_TYPE_ATTR = "original_response_content_type";
            String contentTypeValue = headers.getFirst(HttpHeaders.CONTENT_TYPE);
            if (StringUtils.hasLength(contentTypeValue)) {
               exchange.getAttributes().put(ORIGINAL_RESPONSE_CONTENT_TYPE_ATTR,
                     contentTypeValue);
            }
            
            // 转发状态 , 存在往 GatewayResponse 设置状态
            HttpStatus status = HttpStatus.resolve(res.status().code());
            if (status != null) {
               response.setStatusCode(status);
            }
            else if (response instanceof AbstractServerHttpResponse) {
               ((AbstractServerHttpResponse) response)
                     .setStatusCodeValue(res.status().code());
            }
            else {
               throw new IllegalStateException(
                     "Unable to set status code on response: "
                           + res.status().code() + ", "
                           + response.getClass());
            }

            // 确保 Header filter 在设置状态后运行， 校验 header 中 filter 正常
            HttpHeaders filteredResponseHeaders = HttpHeadersFilter.filter(
                  getHeadersFilters(), headers, exchange, Type.RESPONSE);
            
            //  String TRANSFER_ENCODING = "Transfer-Encoding"
            //  String CONTENT_LENGTH = "Content-Length"
            if (!filteredResponseHeaders
                  .containsKey(HttpHeaders.TRANSFER_ENCODING)
                  && filteredResponseHeaders
                        .containsKey(HttpHeaders.CONTENT_LENGTH)) {
               // content-length 存在需要去掉 Transfer-Encoding
               response.getHeaders().remove(HttpHeaders.TRANSFER_ENCODING);
            }

            exchange.getAttributes().put(CLIENT_RESPONSE_HEADER_NAMES,
                  filteredResponseHeaders.keySet());

            response.getHeaders().putAll(filteredResponseHeaders);

            // 延迟提交响应，直到所有路由过滤器都运行
            // 将客户端响应作为ServerWebExchange属性，稍后写入响应NettyWriteResponseFilter
            exchange.getAttributes().put(CLIENT_RESPONSE_ATTR, res);
            exchange.getAttributes().put(CLIENT_RESPONSE_CONN_ATTR, connection);

            return Mono.just(res);
         });

   if (properties.getResponseTimeout() != null) {
      // 超时异常处理
      responseFlux = responseFlux.timeout(properties.getResponseTimeout(),
            Mono.error(new TimeoutException("Response took longer than timeout: "
                  + properties.getResponseTimeout())))
            .onErrorMap(TimeoutException.class,
                  // GATEWAY_TIMEOUT(504, "Gateway Timeout")
                  th -> new ResponseStatusException(HttpStatus.GATEWAY_TIMEOUT,
                        th.getMessage(), th));
   }

   return responseFlux.then(chain.filter(exchange));
}

java复制代码C- NettyWriteResponseFilter
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
   return chain.filter(exchange).then(Mono.defer(() -> {
      // Step 1 : 获取 GatewayRequest
      Connection connection = exchange.getAttribute(CLIENT_RESPONSE_CONN_ATTR);
      // 连接不存在直接返回空 
      if (connection == null) {
         return Mono.empty();
      }
      
      // Step 2 : 获取 GatewayResponse
      ServerHttpResponse response = exchange.getResponse();
      NettyDataBufferFactory factory = (NettyDataBufferFactory) response
            .bufferFactory();

      // 此处主要包含一个 byteBufflux
      final Flux<NettyDataBuffer> body = connection.inbound().receive().retain()
            .map(factory::wrap);

      // 媒体类型  
      MediaType contentType = null;
      try {
         contentType = response.getHeaders().getContentType();
      }
      catch (Exception e) {
         log.trace("invalid media type", e);
      }
      return (isStreamingMediaType(contentType)
            ? response.writeAndFlushWith(body.map(Flux::just))
            : response.writeWith(body));
   }));
}

xml复制代码<?xml version="1.0" encoding="UTF-8" ?>
<!-- 
    scan:当此属性设置为true时，配置文件如果发生改变，将会被重新加载，默认值为true
    scanPeriod:设置监测配置文件是否有修改的时间间隔，如果没有给出时间单位默认单位是毫秒，当scan为true时此属性生效，默认时间间隔为1分钟
    debug:当此属性设置为true时，将打印出logback内部日志信息，实时查看logback运行状态，默认值为false
 -->
<configuration scan="true" scanPeriod="2 seconds">
    <!--
     	定义滚动记录文件appender 作用:滚动记录文件,先将日志记录到指定文件,当符合某个条件时,将日志记录到其他文件 
        RollingFileAppender class="ch.qos.logback.core.rolling.RollingFileAppender"
       	 参数： 
            <append>:如果是true日志被追加到文件结尾，如果是false清空现存文件,默认是true
            <file>:被写入的文件名，可以是相对目录也可以是绝对目录，如果上级目录不存在会自动创建，没有默认值
            <rollingPolicy>:当发生滚动时，决定RollingFileAppender的行为，涉及文件移动和重命名
            <triggeringPolicy>:告知RollingFileAppender合适激活滚动
            <prudent>:当为true时不支持FixedWindowRollingPolicy支持TimeBasedRollingPolicy，但是有两个限制:1不支持也不允许文件压缩,2不能设置file属性必须留空
    -->
    <appender name="fileAppender" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <!-- 如果是true,日志被追加到文件结尾，如果是false,清空现存文件.默认是true -->
        <prudent>true</prudent>
        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
        	<!-- 每天滚动一次的日志 只保留30天内的日志文件 -->
            <fileNamePattern>logs/%d{yyyy-MM-dd}/springboot_%i.log</fileNamePattern>
            <maxHistory>30</maxHistory>
            <timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
                <maxFileSize>10MB</maxFileSize>
            </timeBasedFileNamingAndTriggeringPolicy>
        </rollingPolicy>
        <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
        	<!-- 对日志进行格式化 -->
            <pattern>%date %level [%thread] %logger{10}.%class{0}#%method[%file:%line] %n%msg%n</pattern>
        	<charset>utf-8</charset>
        </encoder>
    </appender>

    <appender name="errorAppender" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <prudent>true</prudent>
        <rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
            <fileNamePattern>logs/%d{yyyy-MM-dd}/springboot-error_%i.log</fileNamePattern>
            <maxHistory>30</maxHistory>
            <timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
                <maxFileSize>10MB</maxFileSize>
            </timeBasedFileNamingAndTriggeringPolicy>
        </rollingPolicy>
        <encoder>
            <pattern>%date %level [%thread] %logger{10}.%class{0}#%method[%file:%line] %n%msg%n</pattern>
        	<charset>utf-8</charset>
        </encoder>
        <!-- 
             配置日志级别过滤器 作用:根据日志级别进行过滤，如果日志级别等于配置级别过滤器会根据onMath和onMismatch接收或拒绝日志
             参数:
             <level>:设置过滤级别
             <onMatch>:用于配置符合过滤条件的操作
             <onMismatch>:用于配置不符合过滤条件的操作
                                       此处配置为只接收ERROR日志级别信息
         -->
        <filter class="ch.qos.logback.classic.filter.LevelFilter">
            <level>ERROR</level>
            <onMatch>ACCEPT</onMatch>
            <onMismatch>DENY</onMismatch>
        </filter>
    </appender>

    <!-- 定义控制台appender 作用:把日志输出到控制台 class="ch.qos.logback.core.ConsoleAppender" -->
    <appender name="console" class="ch.qos.logback.core.ConsoleAppender">
        <layout class="ch.qos.logback.classic.PatternLayout">
            <pattern>%date %level [%thread] %logger{10}.%class{0}#%method[%file:%line] %n%msg%n</pattern>
        </layout>
    </appender>

    <!-- 将root的打印级别设置为"error",指定了名字为"console","fileAppender","errorAppender"的appender -->
    <root level="error">
    	<appender-ref ref="console"/>
        <appender-ref ref="fileAppender"/>
        <appender-ref ref="errorAppender"/>
    </root>
    
    <!--                    
        logger用来设置某一个包的日志打印级别
        <loger> 仅有一个name属性,一个可选的level和一个可选的addtivity属性
                name:用来指定受此loger约束的某一个包或者具体的某一个类
                level:用来设置打印级别,大小写无关:TRACE, DEBUG, INFO, WARN, ERROR, ALL 和 OFF
                addtivity:是否向上级loger传递打印信息。默认是true,会将信息输入到root配置指定的地方,可以包含多个appender-ref，标识这个appender会添加到这个logger
    -->
    <logger name="com.xcbeyond.springboot" level="debug"/>
</configuration>

arduino复制代码package com.xcbeyond.springboot;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
/**
 * SpringBoot启动类
 * @author xcbeyond
 * 2018年7月2日下午5:41:45
 */
@SpringBootApplication
public class SpringbootApplication {
	private static Logger logger = LoggerFactory.getLogger(SpringbootApplication.class);

	public static void main(String[] args) {
		if(logger.isDebugEnabled()) {
			logger.debug("SpringBoot starting...");
		}
		SpringApplication.run(SpringbootApplication.class, args);
	}
}
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
/**
 * SpringBoot启动类
 * @author xcbeyond
 * 2018年7月2日下午5:41:45
 */
@SpringBootApplication
public class SpringbootApplication {
	private static Logger logger = LoggerFactory.getLogger(SpringbootApplication.class);

	public static void main(String[] args) {
		if(logger.isDebugEnabled()) {
			logger.debug("SpringBoot starting...");
		}
		SpringApplication.run(SpringbootApplication.class, args);
	}
}

shell复制代码# docker images
REPOSITORY            TAG       IMAGE ID       CREATED          SIZE
xmtubuntu             latest    c3e95388a66b   38 seconds ago   114MB

shell复制代码# docker run -it -P --name ubuntu2 xmtubuntu

shell复制代码# docker exec -it ubuntu1 ping 172.18.0.3
PING 172.18.0.3 (172.18.0.3) 56(84) bytes of data.
64 bytes from 172.18.0.3: icmp_seq=1 ttl=64 time=0.071 ms
64 bytes from 172.18.0.3: icmp_seq=2 ttl=64 time=0.070 ms
64 bytes from 172.18.0.3: icmp_seq=3 ttl=64 time=0.077 ms

shell复制代码# docker exec -it ubuntu3 ping ubuntu2
PING ubuntu2 (172.18.0.3) 56(84) bytes of data.
64 bytes from ubuntu2 (172.18.0.3): icmp_seq=1 ttl=64 time=0.093 ms
64 bytes from ubuntu2 (172.18.0.3): icmp_seq=2 ttl=64 time=0.085 ms
64 bytes from ubuntu2 (172.18.0.3): icmp_seq=3 ttl=64 time=0.092 ms
64 bytes from ubuntu2 (172.18.0.3): icmp_seq=4 ttl=64 time=0.073 ms

shell复制代码# docker exec -it ubuntu2 ping ubuntu3
ping: ubuntu3: Name or service not known

shell复制代码:~# docker network ls
NETWORK ID     NAME      DRIVER    SCOPE
8317183dfc58   bridge    bridge    local
997107487c6b   host      host      local
ab130876cbe6   none      null      local

shell复制代码# docker network create --driver bridge --subnet 192.168.0.0/16 --gateway 192.168.0.1 mynet
9a597fc31f1964d434181907e21ff7010738f3f7dc35ba86bf7434f05a6afc4a

shell复制代码# docker run -it -P --name ubuntu1 --net mynet xmtubuntu
# docker run -it -P --name ubuntu2 --net mynet xmtubuntu

shell复制代码# docker network connect -h
Flag shorthand -h has been deprecated, please use --help

Usage:  docker network connect [OPTIONS] NETWORK CONTAINER

Connect a container to a network

shell复制代码# docker network inspect mynet

csharp复制代码public class ThreadLocalUtils {
​
    private static ThreadLocal<Integer> threadLocal = new ThreadLocal<>();
​
    public static void set(Integer value) {
        threadLocal.set(value);
    }
​
    public static Integer get() {
        return Objects.nonNull(threadLocal.get()) ? threadLocal.get() : null;
    }
​
    public static void remove() {
        threadLocal.remove();
    }
}

js复制代码<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>

        <!-- junjit -->
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
            <scope>test</scope>
        </dependency>

js复制代码<?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.3.0.RELEASE</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.example</groupId>
    <artifactId>junit</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>junit</name>
    <description>Demo project for Spring Boot and MyBatis and Swagger</description>

    <properties>
        <java.version>1.8</java.version>
    </properties>

    <dependencies>
        <!-- commons start -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-lang3</artifactId>
            <version>3.7</version>
        </dependency>
        <!-- commons end -->

        <!-- mybatis start-->
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <scope>runtime</scope>
        </dependency>
        <!-- mybatis-spring-boot-starter -->
        <dependency>
            <groupId>org.mybatis.spring.boot</groupId>
            <artifactId>mybatis-spring-boot-starter</artifactId>
            <version>2.1.1</version>
        </dependency>
        <!-- druid -->
        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>druid</artifactId>
            <version>1.1.11</version>
        </dependency>
        <!-- mybatis end-->

        <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>

        <!-- junjit -->
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
            <scope>test</scope>
        </dependency>
    </dependencies>

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

js复制代码package com.example.demo;

import com.example.demo.controller.TestController;
import com.example.demo.service.UserService;
import org.junit.jupiter.api.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.annotation.Rollback;
import org.springframework.test.context.junit4.SpringRunner;

@SpringBootTest(classes = DemoJunitApplication.class)
@RunWith(SpringRunner.class)
@Rollback
@AutoConfigureMockMvc
class DemoApplicationTests {

    @Autowired
    private TestController testController;

    @Autowired
    private UserService userService;

    /**
     * @ClassName 测试接口
     * @Description: 获取所有用户
     * @Author JavaZhan @公众号:Java全栈架构师
     * @Date 2020/6/13
     * @Version V1.0
     **/
    @Test
    void getTest() {
        System.out.println(testController.getTest());

    }


    /**
     * @ClassName getAllUser
     * @Description: 获取所有用户
     * @Author JavaZhan @公众号:Java全栈架构师
     * @Date 2020/6/13
     * @Version V1.0
     **/
    @Test
    void getAllUser() {
        userService.getAllUser().forEach(user -> System.out.println(user));

    }

}

shell复制代码bin/sqoop import \
--connect jdbc:mysql://10.0.10.118:3306/sqoop_test \
--username root \
--password 123456 \
--table sqoop_test \

shell复制代码bin/sqoop import \
--connect jdbc:mysql://10.0.10.118:3306/sqoop_test \
--username root \
--password 123456 \
--table sqoop_test \
--target-dir /user/company \
--delete-target-dir \
--num-mappers 1 \
--fields-terminated-by "\t"

shell复制代码bin/sqoop import \
--connect jdbc:mysql://10.0.10.118:3306/sqoop_test \
--username root \
--password 123456 \
--target-dir /user/company \
--delete-target-dir \
--num-mappers 1 \
--fields-terminated-by "\t" \
--query 'select id,name,sex from sqoop_test where id>2 and $CONDITIONS'

shell复制代码bin/sqoop import \
--connect jdbc:mysql://10.0.10.118:3306/sqoop_test \
--username root \
--password 123456 \
--table sqoop_test \
--where "id<=2" \
--target-dir /user/company \
--delete-target-dir \
--num-mappers 1 \
--fields-terminated-by "\t"

shell复制代码bin/sqoop import \
--connect jdbc:mysql://10.0.10.118:3306/sqoop_test \
--username root \
--password 123456 \
--table sqoop_test \
--where "id>=2" \
--columns name,sex \
--target-dir /user/company \
--delete-target-dir \
--num-mappers 1 \
--fields-terminated-by "\t"

shell复制代码bin/sqoop import \
--connect jdbc:mysql://10.0.10.118:3306/sqoop_test \
--username root \
--password 123456 \
--table sqoop_test \
--incremental append \
--check-column id \
--last-value 5 \
--target-dir /user/company \
--num-mappers 1 \
--fields-terminated-by "\t"

shell复制代码sqoop import \
--connect jdbc:mysql://10.0.10.118:3306/sqoop_test \
--username root \
--password 123456 \
--table sqoop_test \
--incremental lastmodified \
--check-column date \
--last-value "2021-07-26 11:53:50" \
--target-dir /user/company \
--num-mappers 1 \
--fields-terminated-by "\t" \
--merge-key id

shell复制代码--incremental lastmodified \
--check-column date_col \
--last-value "2021-08-02 17:17:23"

shell复制代码bin/sqoop import \
--connect jdbc:mysql://10.0.10.118:3306/sqoop_test \
--username root \
--password 123456 \
--table sqoop_test \
--num-mappers 1 \
--hive-import \
--fields-terminated-by "\t" \
--hive-overwrite \
--hive-table sqoop_test_hive

shell复制代码sqoop import ... --null-string '\\N' --null-non-string '\\N'

shell复制代码create 'sqoop_test_hbase','info'

shell复制代码bin/sqoop import \
--connect jdbc:mysql://10.0.10.118:3306/sqoop_test \
--username root \
--password 123456 \
--table sqoop_test \
--num-mappers 1 \
--column-family "info" \
--hbase-create-table \
--hbase-row-key "id" \
--hbase-table "sqoop_test_hbase" \
--split-by id

shell复制代码$ sqoop export (generic-args) (export-args)
$ sqoop-export (generic-args) (export-args)

shell复制代码bin/sqoop export \
--connect jdbc:mysql://10.0.10.118:3306/sqoop_test \
--username root \
--password 123456 \
--table sqoop_test \
--num-mappers 1 \
--export-dir /user/hive/warehouse/sqoop_test_hive \
--input-fields-terminated-by "\t"

shell复制代码vim test.txt

tex复制代码1,张三,男,2021-08-03 15:05:51
2,李四,女,2021-08-02 15:05:51
3,赵六,男,2021-08-01 15:05:51

shell复制代码sqoop export \
--connect jdbc:mysql://10.0.10.118:3306/sqoop_test \
--username root \
--password 123456 \
--table sqoop_test \
--num-mappers 1 \
--export-dir /user/test.txt \
--input-fields-terminated-by ","

shell复制代码vim test1.txt

shell复制代码1,张三,未知,2021-08-03 15:05:51
2,李四,女,2021-08-02 15:05:51
3,赵六,男,2021-08-01 15:05:51
4,王五,男,2021-07-01 15:05:51

shell复制代码sqoop export \
--connect jdbc:mysql://10.0.10.118:3306/sqoop_test \
--username root \
--password 123456 \
--table sqoop_test \
--num-mappers 1 \
--export-dir /user/test1.txt \
--input-fields-terminated-by "," \
--update-key id \
--update-mode updateonly

shell复制代码vim test3.txt

tex复制代码1,张三,女,2021-08-03 15:05:51
2,李四,女,2021-08-02 15:05:51
3,赵六,男,2021-08-01 15:05:51
4,王五,男,2021-07-01 15:05:51

shell复制代码sudo -u hdfs hdfs dfs -put ./test3.txt /user/

shell复制代码sqoop export \
--connect jdbc:mysql://10.0.10.118:3306/sqoop_test \
--username root \
--password 123456 \
--table sqoop_test \
--num-mappers 1 \
--export-dir /user/test3.txt \
--input-fields-terminated-by "," \
--update-key id \
--update-mode allowinsert

shell复制代码vim sqp_export.opt

shell复制代码export
--connect
jdbc:mysql://10.0.10.118:3306/sqoop_test
--username
root
--password
123456
--table
sqoop_test
--num-mappers
1
--export-dir
/user/hive/warehouse/sqoop_test_hive
--input-fields-terminated-by
"\t"

shell复制代码sudo -u hdfs sqoop --options-file ./sqp_export.opt

lua复制代码-- Classic hello program.

print("helloworld")



-------- Output ------

hello

lua复制代码-- Single line comments in Lua start with double hyphen.

--[[ Multiple line comments start
with double hyphen and two square brackets.
  and end with two square brackets. ]]

-- And of course this example produces no
-- output, since it's all comments!

lua复制代码-- Variables hold values which have types, variables don't have types.

a=1
b="abc"
c={}
d=print

print(type(a))
print(type(b))
print(type(c))
print(type(d))


-------- Output ------

number
string
table
function

lua复制代码-- Variable names consist of letters, digits and underscores.
-- They cannot start with a digit.

one_two_3 = 123 -- is valid varable name

-- 1_two_3 is not a valid variable name.

lua复制代码-- The underscore is typically used to start special values
-- like _VERSION in Lua.

print(_VERSION)

-- So don't use variables that start with _,
-- but a single underscore _ is often used as a
-- dummy variable.


-------- Output ------

Lua 5.1

lua复制代码-- Lua is case sensitive so all variable names & keywords
-- must be in correct case.

ab=1
Ab=2
AB=3
print(ab,Ab,AB)



-------- Output ------

1       2       3

lua复制代码-- Lua reserved words are: and, break, do, else, elseif,
--     end, false, for, function, if, in, local, nil, not, or,
--     repeat, return, then, true, until, while.

-- Keywords cannot be used for variable names,
-- 'and' is a keyword, but AND is not, so it is a legal variable name.
AND=3
print(AND)



-------- Output ------

3

lua复制代码a="single 'quoted' string and double \"quoted\" string inside"
b='single \'quoted\' string and double "quoted" string inside'
c= [[ multiple line
with 'single'
and "double" quoted strings inside.]]

print(a)
print(b)
print(c)


-------- Output ------

single 'quoted' string and double "quoted" string inside
single 'quoted' string and double "quoted" string inside
 multiple line
with 'single'
and "double" quoted strings inside.

lua复制代码-- a,b = b,a is valid
a,b,c,d,e = 1,2,3,'four','five'
print(a,b,c,d,e)

-------- Output ------
1       2       3       four    five

lua复制代码-- Multiple assignments allows one line to swap two variables.

print(a,b)
a,b=b,a
print(a,b)

-------- Output ------

1       2
2       1

lua复制代码-- Multiple assignment showing different number formats.
-- Two dots (..) are used to concatenate strings (or a
-- string and a number).

a,b,c,d,e = 1, 1.123, 1E9, -123, .0008
print("a="..a, "b="..b, "c="..c, "d="..d, "e="..e)



-------- Output ------

a=1     b=1.123 c=1000000000    d=-123  e=0.0008

lua复制代码-- More writing output.

print "Hello from Lua!"
print("Hello from Lua!")



-------- Output ------

Hello from Lua!
Hello from Lua!

lua复制代码-- io.write writes to stdout but without new line.

io.write("Hello from Lua!")
io.write("Hello from Lua!")

-- Use an empty print to write a single new line.
print()



-------- Output ------

Hello from Lua!Hello from Lua!

lua复制代码-- Simple table creation.

a={} -- {} creates an empty table
b={1,2,3} -- creates a table containing numbers 1,2,3
c={"a","b","c"} -- creates a table containing strings a,b,c
print(a,b,c) -- tables don't print directly, we'll get back to this!!


-------- Output ------

table: 008A48A8 table: 008A4420 table: 008A4768


--在 Lua 里表的默认初始索引一般以 1 开始。
> b = {4,5,6,7,8}
> print(b)
table: 009B9978
> print(b[0])
nil
> print(b[1])
4
> print(b[2])
5

lua复制代码-- Associate index style.

address={} -- empty address
address.Street="Wyman Street"
address.StreetNumber=360
address.AptNumber="2a"
address.City="Watertown"
address.State="Vermont"
address.Country="USA"

print(address.StreetNumber, address["AptNumber"])



-------- Output ------

360     2a

lua复制代码-- Simple if.

a=1
if a==1 then
    print ("a is one")
end


-------- Output ------

a is one

lua复制代码b="happy"
if b=="sad" then
    print("b is sad")
else
    print("b is not sad")
end


-------- Output ------

b is not sad

lua复制代码c=3
if c==1 then
    print("c is 1")
elseif c==2 then
    print("c is 2")
else
    print("c isn't 1 or 2, c is "..tostring(c))
end


-------- Output ------

c isn't 1 or 2, c is 3

lua复制代码-- value = test and x or y

a=1
b=(a==1) and "one" or "not one"
print(b)

-- is equivalent to
a=1
if a==1 then
    b = "one"
else
    b = "not one"
end
print(b)


-------- Output ------

one
one

lua复制代码a=1
while a~=5 do -- Lua uses ~= to mean not equal
    a=a+1
    io.write(a.." ")
end


-------- Output ------

2 3 4 5

lua复制代码a=0
repeat
    a=a+1
    print(a)
until a==5


-------- Output ------

1
2
3
4
5

lua复制代码-- Numeric iteration form.

-- Count from 1 to 4 by 1.
for a=1,4 do io.write(a) end

print()

-- Count from 1 to 6 by 3.
for a=1,6,3 do io.write(a) end


-------- Output ------

1234
14

lua复制代码-- Sequential iteration form.

for key,value in pairs({1,2,3,4}) do print(key, value) end


-------- Output ------

1       1
2       2
3       3
4       4

lua复制代码-- Simple way to print tables. ##  iterator

a={1,2,3,4,"five","elephant", "mouse"}

for i,v in pairs(a) do print(i,v) end


-------- Output ------

1       1
2       2
3       3
4       4
5       five
6       elephant
7       mouse

lua复制代码-- break is used to exit a loop.
a=0
while true do
    a=a+1
    if a==10 then
        break
    end
end

print(a)


-------- Output ------

10

Press 'Enter' key for next example

lua复制代码-- Define a function without parameters or return value.
function myFirstLuaFunction()
    print("My first lua function was called")
end

-- Call myFirstLuaFunction.
myFirstLuaFunction()


-------- Output ------

My first lua function was called

lua复制代码-- Define a function with a return value.
function mySecondLuaFunction()
    return "string from my second function"
end

-- Call function returning a value.
a=mySecondLuaFunction("string")
print(a)

lua复制代码-- Define function with multiple parameters and multiple return values.
function myFirstLuaFunctionWithMultipleReturnValues(a,b,c)
    return a,b,c,"My first lua function with multiple return values", 1, true
end

a,b,c,d,e,f = myFirstLuaFunctionWithMultipleReturnValues(1,2,"three")
print(a,b,c,d,e,f)


-------- Output ------

1       2       three   My first lua function with multiple return values 1true

lua复制代码-- All variables are global in scope by default.

b="global"

-- To make local variables you must put the keyword 'local' in front.
function myfunc()
    local b=" local variable"
    a="global variable"
    print(a,b)
end

myfunc()
print(a,b)


-------- Output ------

global variable  local variable
global variable global

lua复制代码-- An implementation of printf.

function printf(fmt, ...)
    io.write(string.format(fmt, ...))
end

printf("Hello %s from %s on %s\n",
       os.getenv"USER" or "there", _VERSION, os.date())


-------- Output ------

Hello there from Lua 5.1 on 08/05/19 09:34:49

lua复制代码--[[

 Standard Libraries

  Lua has standard built-in libraries for common operations in
  math, string, table, input/output & operating system facilities.

 External Libraries

  Numerous other libraries have been created: sockets, XML, profiling,
  logging, unittests, GUI toolkits, web frameworks, and many more.

]]



-------- Output ------