go复制代码import (
    "bytes"
    "crypto/aes"
    "fmt"
    "crypto/cipher"
    "encoding/base64"
)

func main() {
    orig := "hello world"
    key := "123456781234567812345678"
    fmt.Println("原文：", orig)

    encryptCode := AesEncrypt(orig, key)
    fmt.Println("密文：" , encryptCode)

    decryptCode := AesDecrypt(encryptCode, key)
    fmt.Println("解密结果：", decryptCode)
}

func AesEncrypt(orig string, key string) string {
    // 转成字节数组
    origData := []byte(orig)
    k := []byte(key)

	// 分组秘钥
	block, err := aes.NewCipher(k)
	if err != nil {
		panic(fmt.Sprintf("key 长度必须 16/24/32长度: %s", err.Error()))
	}
    // 获取秘钥块的长度
    blockSize := block.BlockSize()
    // 补全码
    origData = PKCS7Padding(origData, blockSize)
    // 加密模式
    blockMode := cipher.NewCBCEncrypter(block, k[:blockSize])
    // 创建数组
    cryted := make([]byte, len(origData))
    // 加密
    blockMode.CryptBlocks(cryted, origData)
    //使用RawURLEncoding 不要使用StdEncoding
    //不要使用StdEncoding  放在url参数中回导致错误
    return base64.RawURLEncoding.EncodeToString(cryted)

}

func AesDecrypt(cryted string, key string) string {
    //使用RawURLEncoding 不要使用StdEncoding
    //不要使用StdEncoding  放在url参数中回导致错误
    crytedByte, _ := base64.RawURLEncoding.DecodeString(cryted)
    k := []byte(key)

	// 分组秘钥
	block, err := aes.NewCipher(k)
	if err != nil {
		panic(fmt.Sprintf("key 长度必须 16/24/32长度: %s", err.Error()))
	}
    // 获取秘钥块的长度
    blockSize := block.BlockSize()
    // 加密模式
    blockMode := cipher.NewCBCDecrypter(block, k[:blockSize])
    // 创建数组
    orig := make([]byte, len(crytedByte))
    // 解密
    blockMode.CryptBlocks(orig, crytedByte)
    // 去补全码
    orig = PKCS7UnPadding(orig)
    return string(orig)
}

//补码
func PKCS7Padding(ciphertext []byte, blocksize int) []byte {
    padding := blocksize - len(ciphertext)%blocksize
    padtext := bytes.Repeat([]byte{byte(padding)}, padding)
    return append(ciphertext, padtext...)
}

//去码
func PKCS7UnPadding(origData []byte) []byte {
    length := len(origData)
    unpadding := int(origData[length-1])
    return origData[:(length - unpadding)]
}

go复制代码func ZeroPadding(ciphertext []byte, blockSize int) []byte {
    padding := blockSize - len(ciphertext)%blockSize
    padtext := bytes.Repeat([]byte{0}, padding)
    return append(ciphertext, padtext...)
}

func ZeroUnPadding(origData []byte) []byte {
    return bytes.TrimFunc(origData,
        func(r rune) bool {
            return r == rune(0)
        })
}

go复制代码func Encrypt(text string, key []byte) (string, error) {
    src := []byte(text)
    block, err := des.NewCipher(key)
    if err != nil {
        return "", err
    }
    bs := block.BlockSize()
    src = ZeroPadding(src, bs)
    if len(src)%bs != 0 {
        return "", errors.New("Need a multiple of the blocksize")
    }
    out := make([]byte, len(src))
    dst := out
    for len(src) > 0 {
        block.Encrypt(dst, src[:bs])
        src = src[bs:]
        dst = dst[bs:]
    }
    return hex.EncodeToString(out), nil
}

go复制代码func Decrypt(decrypted string , key []byte) (string, error) {
    src, err := hex.DecodeString(decrypted)
    if err != nil {
        return "", err
    }
    block, err := des.NewCipher(key)
    if err != nil {
        return "", err
    }
    out := make([]byte, len(src))
    dst := out
    bs := block.BlockSize()
    if len(src)%bs != 0 {
        return "", errors.New("crypto/cipher: input not full blocks")
    }
    for len(src) > 0 {
        block.Decrypt(dst, src[:bs])
        src = src[bs:]
        dst = dst[bs:]
    }
    out = ZeroUnPadding(out)
    return string(out), nil
}

go复制代码func main() {
    key := []byte("2fa6c1e9")
    str :="I love this beautiful world!"
    strEncrypted, err := Encrypt(str, key)
    if err != nil {
        log.Fatal(err)
    }
    fmt.Println("Encrypted:", strEncrypted)
    strDecrypted, err := Decrypt(strEncrypted, key)
    if err != nil {
        log.Fatal(err)
    }
    fmt.Println("Decrypted:", strDecrypted)
}
//Output:
//Encrypted: 5d2333b9fbbe5892379e6bcc25ffd1f3a51b6ffe4dc7af62beb28e1270d5daa1
//Decrypted: I love this beautiful world!

go复制代码import (
    "crypto/rand"
    "crypto/rsa"
    "crypto/x509"
    "encoding/base64"
    "encoding/pem"
    "errors"
    "fmt"
)

// 私钥生成
//openssl genrsa -out rsa_private_key.pem 1024
var privateKey = []byte(`
-----BEGIN RSA PRIVATE KEY-----
MIICWwIBAAKBgQDcGsUIIAINHfRTdMmgGwLrjzfMNSrtgIf4EGsNaYwmC1GjF/bM
h0Mcm10oLhNrKNYCTTQVGGIxuc5heKd1gOzb7bdTnCDPPZ7oV7p1B9Pud+6zPaco
qDz2M24vHFWYY2FbIIJh8fHhKcfXNXOLovdVBE7Zy682X1+R1lRK8D+vmQIDAQAB
AoGAeWAZvz1HZExca5k/hpbeqV+0+VtobMgwMs96+U53BpO/VRzl8Cu3CpNyb7HY
64L9YQ+J5QgpPhqkgIO0dMu/0RIXsmhvr2gcxmKObcqT3JQ6S4rjHTln49I2sYTz
7JEH4TcplKjSjHyq5MhHfA+CV2/AB2BO6G8limu7SheXuvECQQDwOpZrZDeTOOBk
z1vercawd+J9ll/FZYttnrWYTI1sSF1sNfZ7dUXPyYPQFZ0LQ1bhZGmWBZ6a6wd9
R+PKlmJvAkEA6o32c/WEXxW2zeh18sOO4wqUiBYq3L3hFObhcsUAY8jfykQefW8q
yPuuL02jLIajFWd0itjvIrzWnVmoUuXydwJAXGLrvllIVkIlah+lATprkypH3Gyc
YFnxCTNkOzIVoXMjGp6WMFylgIfLPZdSUiaPnxby1FNM7987fh7Lp/m12QJAK9iL
2JNtwkSR3p305oOuAz0oFORn8MnB+KFMRaMT9pNHWk0vke0lB1sc7ZTKyvkEJW0o
eQgic9DvIYzwDUcU8wJAIkKROzuzLi9AvLnLUrSdI6998lmeYO9x7pwZPukz3era
zncjRK3pbVkv0KrKfczuJiRlZ7dUzVO0b6QJr8TRAA==
-----END RSA PRIVATE KEY-----
`)

// 公钥: 根据私钥生成
//openssl rsa -in rsa_private_key.pem -pubout -out rsa_public_key.pem
var publicKey = []byte(`
-----BEGIN PUBLIC KEY-----
MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQDcGsUIIAINHfRTdMmgGwLrjzfM
NSrtgIf4EGsNaYwmC1GjF/bMh0Mcm10oLhNrKNYCTTQVGGIxuc5heKd1gOzb7bdT
nCDPPZ7oV7p1B9Pud+6zPacoqDz2M24vHFWYY2FbIIJh8fHhKcfXNXOLovdVBE7Z
y682X1+R1lRK8D+vmQIDAQAB
-----END PUBLIC KEY-----
`)

// 加密
func RsaEncrypt(origData []byte) ([]byte, error) {
    //解密pem格式的公钥
    block, _ := pem.Decode(publicKey)
    if block == nil {
        return nil, errors.New("public key error")
    }
    // 解析公钥
    pubInterface, err := x509.ParsePKIXPublicKey(block.Bytes)
    if err != nil {
        return nil, err
    }
    // 类型断言
    pub := pubInterface.(*rsa.PublicKey)
    //加密
    return rsa.EncryptPKCS1v15(rand.Reader, pub, origData)
}

// 解密
func RsaDecrypt(ciphertext []byte) ([]byte, error) {
    //解密
    block, _ := pem.Decode(privateKey)
    if block == nil {
        return nil, errors.New("private key error!")
    }
    //解析PKCS1格式的私钥
    priv, err := x509.ParsePKCS1PrivateKey(block.Bytes)
    if err != nil {
        return nil, err
    }
    // 解密
    return rsa.DecryptPKCS1v15(rand.Reader, priv, ciphertext)
}
func main() {
    data, _ := RsaEncrypt([]byte("hello world"))
    fmt.Println(base64.StdEncoding.EncodeToString(data))
    origData, _ := RsaDecrypt(data)
    fmt.Println(string(origData))
}

go复制代码import (
	"crypto/ecdsa"
	"crypto/elliptic"
	"crypto/rand"
	"crypto/x509"
	"encoding/pem"
	"github.com/spf13/cobra"
	"log"
	"os"
)


// ecdsaCmd represents the doc command
 func keyPairs(keyName string) {
    //elliptic.P256(),elliptic.P384(),elliptic.P521()

    privateKey, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
    if err != nil {
        log.Fatal(err)
    }
    x509Encoded, _ := x509.MarshalECPrivateKey(privateKey)
    privateBs := pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: x509Encoded})
    privateFile, err := os.Create(keyName + ".private.pem")
    if err != nil {
        log.Fatal(err)
    }
    _, err = privateFile.Write(privateBs)
    if err != nil {
        log.Fatal(err)
    }
    x509EncodedPub, _ := x509.MarshalPKIXPublicKey(privateKey.Public())
    publicBs := pem.EncodeToMemory(&pem.Block{Type: "PUBLIC KEY", Bytes: x509EncodedPub})
    publicKeyFile, err := os.Create(keyName + ".public.pem")
    if err != nil {
        log.Fatal(err)
    }
    _, err = publicKeyFile.Write(publicBs)
    if err != nil {
        log.Fatal(err)
    }
}

go复制代码func Check(content, encrypted string) bool {
    return strings.EqualFold(Encode(content), encrypted)
}
func Encode(data string) string {
    h := md5.New()
    h.Write([]byte(data))
    return hex.EncodeToString(h.Sum(nil))
}

go复制代码func main() {
     strTest := "I love this beautiful world!"
    strEncrypted := "98b4fc4538115c4980a8b859ff3d27e1"
    fmt.Println(Check(strTest, strEncrypted))
}
//Output:
//true

Sha1

package main

import (
	"crypto/sha1"
	 "fmt"
)
func main() {
   s := "sha1 this string"
   //产生一个散列值得方式是 sha1.New(),sha1.Write(bytes),然后 sha1.Sum([]byte{}).这里我们从一个新的散列开始.
   h := sha1.New()
   //写入要处理的字节.如果是一个字符串,需要使用[]byte(s) 来强制转换成字节数组.
   h.Write([]byte(s))
   //这个用来得到最终的散列值的字符切片.Sum 的参数可以用来都现有的字符切片追加额外的字节切片：一般不需要要.
   bs := h.Sum(nil)
   //SHA1 值经常以 16 进制输出,例如在 git commit 中.使用%x 来将散列结果格式化为 16 进制字符串.
   fmt.Println(s)
   fmt.Printf("%x\n", bs)
}

go复制代码package main

import (
    "crypto/sha1"
    "fmt"
)

func main() {
    s := "sha1 this string"

    h := sha1.New()

    h.Write([]byte(s))

    bs := h.Sum(nil)

    fmt.Println(s)
    fmt.Printf("%x\n", bs)
}

go复制代码package main

import (
    "crypto/hmac"
    "crypto/sha256"
    "encoding/hex"
    "fmt"
)

func main() {

    secret := "mysecret"
    data := "data"
    fmt.Printf("Secret: %s Data: %s\n", secret, data)

    // Create a new HMAC by defining the hash type and the key (as byte array)
    h := hmac.New(sha256.New, []byte(secret))

    // Write Data to it
    h.Write([]byte(data))

    // Get result and encode as hexadecimal string
    sha := hex.EncodeToString(h.Sum(nil))

    fmt.Println("Result: " + sha)
}

go复制代码const (
   base64Table = "IJjkKLMNO567PQX12RVW3YZaDEFGbcdefghiABCHlSTUmnopqrxyz04stuvw89+/"

)

var coder = base64.NewEncoding(base64Table)

func Base64Encode(src []byte) []byte {         //编码
   return []byte(coder.EncodeToString(src))
}

func Base64Decode(src []byte) ([]byte, error) {   //解码
   return coder.DecodeString(string(src))
}

cs复制代码public void Main()
{
        // TODO: Add your code here
        var sTaskName = Dts.Variables["TaskName"].Value.ToString();
        var boxRes = MessageBox.Show("设置MyBool为True?", sTaskName, MessageBoxButtons.YesNo);
        if (boxRes == DialogResult.Yes)
        {
            Dts.Variables["User::MyBool"].Value = true;
        }
        else
        {
            Dts.Variables["User::MyBool"].Value = false;
        }
        Dts.TaskResult = (int)ScriptResults.Success;
}

cs复制代码public void Main()
{
         // TODO: Add your code here
         var sTaskName = Dts.Variables["TaskName"].Value.ToString();
         var boxRes = MessageBox.Show("set "+sTaskName + " Succeed?", "确认", MessageBoxButtons.YesNo);
         if (boxRes == DialogResult.Yes)
         {
             Dts.TaskResult = (int)ScriptResults.Success;
         }
         else
         {
             Dts.TaskResult = (int)ScriptResults.Failure;
         }
}

c复制代码dummyUser@XPS13:~$ redis-cli -h 10.10.10.1 
10.10.10.1:6379> config set dir /usr/local/nginx/html 
OK 
10.10.10.1:6379> config set dbfilename redis.php 
OK 
10.10.10.1:6379> set test "<?php phpinfo(); ?>" 
OK 
10.10.10.1:6379> save 
OK

C复制代码dummyUser@XPS13:~$ redis-cli -h 10.10.10.1 
10.10.10.1:6379> config set dir /root/.ssh 
OK 
10.10.10.1:6379> config set dbfilename authorized_keys 
OK 
10.10.10.1:6379> save 
OK

C复制代码dummyUser@XPS13:~$ echo -e "\n\n * * * * * command to execute" | redis-cli -h 10.10.10.1 -x set cron 
OK 
dummyUser@XPS13:~$ redis-cli -h 10.10.10.1 
10.10.10.1:6379> config set dir /var/spool/cron 
OK 
10.10.10.1:6379> config set dbfilename root 
OK 
10.10.10.1:6379> save 
OK

text复制代码rename-command FLUSHDB "" 
rename-command CONFIG "" 
rename-command KEYS ""

json复制代码spring:
  cloud:
		#SCG的配置，对应GatewayProperties
    gateway:
      routes:
      - id: user-service #路由的编号(唯一)
        uri: http://127.0.0.1:8080 #路由到的目标地址
        predicates: # 断言，作为路由的匹配条件 对应RouteDefinition，可以配置多个
        	- Path=/login,/loginUser
				filters:
					- StripPrefix=1 #下边说

json复制代码# Auto Configure
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
org.springframework.cloud.gateway.config.GatewayClassPathWarningAutoConfiguration,\
org.springframework.cloud.gateway.config.GatewayAutoConfiguration,\
org.springframework.cloud.gateway.config.GatewayHystrixCircuitBreakerAutoConfiguration,\
org.springframework.cloud.gateway.config.GatewayResilience4JCircuitBreakerAutoConfiguration,\
org.springframework.cloud.gateway.config.GatewayLoadBalancerClientAutoConfiguration,\
org.springframework.cloud.gateway.config.GatewayNoLoadBalancerClientAutoConfiguration,\
org.springframework.cloud.gateway.config.GatewayMetricsAutoConfiguration,\
org.springframework.cloud.gateway.config.GatewayRedisAutoConfiguration,\
org.springframework.cloud.gateway.discovery.GatewayDiscoveryClientAutoConfiguration,\
org.springframework.cloud.gateway.config.SimpleUrlHandlerMappingGlobalCorsAutoConfiguration,\
org.springframework.cloud.gateway.config.GatewayReactiveLoadBalancerClientAutoConfiguration

org.springframework.boot.env.EnvironmentPostProcessor=\
org.springframework.cloud.gateway.config.GatewayEnvironmentPostProcessor

json复制代码	/**
	 *  读取配置文件中配置的的{@link RouteDefinition} 
   			->>{@link FilterDefinition} ->> {@link PredicateDefinition}并封装
	 * @return
	 */
	@Bean
	public GatewayProperties gatewayProperties() {
		return new GatewayProperties();
	}

json复制代码public class GatewayProperties {

	/**
	 * Properties prefix.
	 */
	public static final String PREFIX = "spring.cloud.gateway";
	/**
	 * List of Routes.
	 */
	@NotNull
	@Valid
	private List<RouteDefinition> routes = new ArrayList<>();

	/**
	 * 作用于每个路由的过滤器列表
	 */
	private List<FilterDefinition> defaultFilters = new ArrayList<>();
					..........省略部分代码............
}

json复制代码public class RouteDefinition {
	
  //路由ID，如果配置为空则SCG会生成随机生成一个
	private String id;
	
	@NotEmpty
	@Valid
  //配置的断言信息
	private List<PredicateDefinition> predicates = new ArrayList<>();

	@Valid
  //配置的过滤器信息
	private List<FilterDefinition> filters = new ArrayList<>();

	@NotNull
  //需要转发到的目的URI
	private URI uri;
}

json复制代码public class PredicateDefinition {

	@NotNull
	/**
	 * 断言的名称，与{@link AbstractRoutePredicateFactory}的子类名称前缀相同
	 */
	private String name;

	/**
	 * 断言的参数  key:_genkey_0 value:/login
	 */
	private Map<String, String> args = new LinkedHashMap<>();

	public PredicateDefinition() {
	}

	public PredicateDefinition(String text) {
		int eqIdx = text.indexOf('=');
		if (eqIdx <= 0) {
			throw new ValidationException("Unable to parse PredicateDefinition text '"
					+ text + "'" + ", must be of the form name=value");
		}
		setName(text.substring(0, eqIdx));
		//将配置的字符串参数中"="右边的字符串以","分割
		String[] args = tokenizeToStringArray(text.substring(eqIdx + 1), ",");
		//遍历","分割后的结果，随机生成一个key(_genkey_+参数下标)，value为参数
		for (int i = 0; i < args.length; i++) {
			this.args.put(NameUtils.generateName(i), args[i]);
		}
	}
}

json复制代码public interface RouteDefinitionLocator {
  
	//获取所有的路由信息
	Flux<RouteDefinition> getRouteDefinitions();

}

json复制代码/**
	 * PropertiesRouteDefinitionLocator 是{@link RouteDefinitionLocator}的实现类 用于存储从配置文件中读取的路由信息
	 * @param properties 即为上边装配的GatewayProperties Bean
	 * @return
	 */
	@Bean
	@ConditionalOnMissingBean
	public PropertiesRouteDefinitionLocator propertiesRouteDefinitionLocator(
			GatewayProperties properties) {
		return new PropertiesRouteDefinitionLocator(properties);
	}

json复制代码/**
	 * 将上边装配的RouteDefinitionLocator再次进行组装
	 * @param routeDefinitionLocators
	 * @return
	 */
	@Bean
	@Primary //定义为Primary是为了下边装配RouteDefinitionRouteLocator时此Bean为注入的RouteDefinitionLocator Bean
	public RouteDefinitionLocator routeDefinitionLocator(
			List<RouteDefinitionLocator> routeDefinitionLocators) {
		return new CompositeRouteDefinitionLocator(
				Flux.fromIterable(routeDefinitionLocators));
	}

json复制代码public class CompositeRouteDefinitionLocator implements RouteDefinitionLocator {

	@Override
	public Flux<RouteDefinition> getRouteDefinitions() {
		return this.delegates
				.flatMapSequential(RouteDefinitionLocator::getRouteDefinitions)
				.flatMap(routeDefinition -> {
					if (routeDefinition.getId() == null) {
						//如果路由id为空，则生成一个
						return randomId().map(id -> {
							routeDefinition.setId(id);
							if (log.isDebugEnabled()) {
								log.debug(
										"Id set on route definition: " + routeDefinition);
							}
							return routeDefinition;
						});
					}
					return Mono.just(routeDefinition);
				});
	}

	protected Mono<String> randomId() {
		return Mono.fromSupplier(idGenerator::toString)
				.publishOn(Schedulers.boundedElastic());
	}

}

json复制代码@Bean
	@ConditionalOnEnabledFilter
	public AddRequestHeaderGatewayFilterFactory addRequestHeaderGatewayFilterFactory() {
		return new AddRequestHeaderGatewayFilterFactory();
	}

	@Bean
	@ConditionalOnEnabledFilter
	public MapRequestHeaderGatewayFilterFactory mapRequestHeaderGatewayFilterFactory() {
		return new MapRequestHeaderGatewayFilterFactory();
	}
....等等...

json复制代码@Bean
	@ConditionalOnEnabledPredicate
	public AfterRoutePredicateFactory afterRoutePredicateFactory() {
		return new AfterRoutePredicateFactory();
	}

	@Bean
	@ConditionalOnEnabledPredicate
	public BeforeRoutePredicateFactory beforeRoutePredicateFactory() {
		return new BeforeRoutePredicateFactory();
	}
....等等...

json复制代码public interface RouteLocator {
	//用来获取路由的
	Flux<Route> getRoutes();

}

json复制代码	/**
	 *
	 * @param properties 即为装配的GatewayProperties Bean
	 * @param gatewayFilters 即为装配的 GatewayFilterFactory所有的实现类
	 * @param predicates 即为装配的  RoutePredicateFactory 所有的实现类
	 * @param routeDefinitionLocator 即为装配的RouteDefinitionLocator ->CompositeRouteDefinitionLocator
	 * @return
	 */
	@Bean
	public RouteLocator routeDefinitionRouteLocator(GatewayProperties properties,
			List<GatewayFilterFactory> gatewayFilters,
			List<RoutePredicateFactory> predicates,
			RouteDefinitionLocator routeDefinitionLocator,
			ConfigurationService configurationService) {
		return new RouteDefinitionRouteLocator(routeDefinitionLocator, predicates,
				gatewayFilters, properties, configurationService);
	}

/**
	 * CachingRouteLocator为RoutePredicateHandlerMapping的RouteLocator,见 {@link this#routePredicateHandlerMapping}
	 * @param routeLocators 上边装配的RouteDefinitionRouteLocator的Bean
	 * @return
	 */
	@Bean
	@Primary //定义为Primary为下边装配RoutePredicateHandlerMapping时使用
	@ConditionalOnMissingBean(name = "cachedCompositeRouteLocator")
	public RouteLocator cachedCompositeRouteLocator(List<RouteLocator> routeLocators) {
		return new CachingRouteLocator(
				new CompositeRouteLocator(Flux.fromIterable(routeLocators)));
	}

json复制代码public class RouteDefinitionRouteLocator
		implements RouteLocator, BeanFactoryAware, ApplicationEventPublisherAware {

	/**
	 * Default filters name.
	 */
	public static final String DEFAULT_FILTERS = "defaultFilters";

	protected final Log logger = LogFactory.getLog(getClass());

	private final RouteDefinitionLocator routeDefinitionLocator;

	private final Map<String, RoutePredicateFactory> predicates = new LinkedHashMap<>();

	private final Map<String, GatewayFilterFactory> gatewayFilterFactories = new HashMap<>();

	private final GatewayProperties gatewayProperties;

	public RouteDefinitionRouteLocator(RouteDefinitionLocator routeDefinitionLocator,
			List<RoutePredicateFactory> predicates,
			List<GatewayFilterFactory> gatewayFilterFactories,
			GatewayProperties gatewayProperties,
			ConfigurationService configurationService) {
		this.routeDefinitionLocator = routeDefinitionLocator;
		this.configurationService = configurationService;
		//初始化Predicate断言信息（所有的）
		initFactories(predicates);
		//初始化Filter信息（所有的），与初始化Predicate断言信息类似
		gatewayFilterFactories.forEach(
				factory -> this.gatewayFilterFactories.put(factory.name(), factory));
		this.gatewayProperties = gatewayProperties;
	}

	private void initFactories(List<RoutePredicateFactory> predicates) {
		predicates.forEach(factory -> {
			//key为RoutePredicateFactory实现类的名称前缀如AfterRoutePredicateFactory则key为After
			String key = factory.name();
			if (this.predicates.containsKey(key)) {
				this.logger.warn("A RoutePredicateFactory named " + key
						+ " already exists, class: " + this.predicates.get(key)
						+ ". It will be overwritten.");
			}
			//如果已经存在该断言Factory，则覆盖，也就是说以SCG内置的为主
			this.predicates.put(key, factory);
			if (logger.isInfoEnabled()) {
				logger.info("Loaded RoutePredicateFactory [" + key + "]");
			}
		});
	}

json复制代码	@Override
	public Flux<Route> getRoutes() {
		//通过RouteDefinitions获取Route
		Flux<Route> routes = this.routeDefinitionLocator.getRouteDefinitions()
				.map(this::convertToRoute);

		if (!gatewayProperties.isFailOnRouteDefinitionError()) {
			// instead of letting error bubble up, continue
			routes = routes.onErrorContinue((error, obj) -> {
				if (logger.isWarnEnabled()) {
					logger.warn("RouteDefinition id " + ((RouteDefinition) obj).getId()
							+ " will be ignored. Definition has invalid configs, "
							+ error.getMessage());
				}
			});
		}
		
		return routes.map(route -> {
			if (logger.isDebugEnabled()) {
				logger.debug("RouteDefinition matched: " + route.getId());
			}
			return route;
		});
	}

json复制代码/**
	 * 将RouteDefinition转换为Route
	 * @param routeDefinition
	 * @return
	 */
	private Route convertToRoute(RouteDefinition routeDefinition) {
		/**
		 * 重点
		 * 获取RouteDefinition对应的断言
		 */
		AsyncPredicate<ServerWebExchange> predicate = combinePredicates(routeDefinition);
		/**
		 * 重点
		 * 获取RouteDefinition对应的Filter
		 */
		List<GatewayFilter> gatewayFilters = getFilters(routeDefinition);

		return Route.async(routeDefinition).asyncPredicate(predicate)
				.replaceFilters(gatewayFilters).build();
	}

json复制代码private AsyncPredicate<ServerWebExchange> combinePredicates(
			RouteDefinition routeDefinition) {
		List<PredicateDefinition> predicates = routeDefinition.getPredicates();
		if (predicates == null || predicates.isEmpty()) {
			// this is a very rare case, but possible, just match all
			return AsyncPredicate.from(exchange -> true);
		}
		AsyncPredicate<ServerWebExchange> predicate = lookup(routeDefinition,
				predicates.get(0));
		//此处是为了如果此RouteDefinition配置了多个Predicate，将多个AsyncPredicate通过与(and)连接起来
		for (PredicateDefinition andPredicate : predicates.subList(1,
				predicates.size())) {
			AsyncPredicate<ServerWebExchange> found = lookup(routeDefinition,
					andPredicate);
			predicate = predicate.and(found);
		}

		return predicate;
	}

private AsyncPredicate<ServerWebExchange> lookup(RouteDefinition route,
			PredicateDefinition predicate) {
		//查找到PredicateDefinition对应的RoutePredicateFactory
		RoutePredicateFactory<Object> factory = this.predicates.get(predicate.getName());
		if (factory == null) {
			throw new IllegalArgumentException(
					"Unable to find RoutePredicateFactory with name "
							+ predicate.getName());
		}
		if (logger.isDebugEnabled()) {
			logger.debug("RouteDefinition " + route.getId() + " applying "
					+ predicate.getArgs() + " to " + predicate.getName());
		}
		//每个RoutePredicateFactory实现中都有Config，可以理解为我们配置的参数规则，生成此Config
		// @formatter:off
		Object config = this.configurationService.with(factory)
				.name(predicate.getName())
				.properties(predicate.getArgs())
				.eventFunction((bound, properties) -> new PredicateArgsEvent(
						RouteDefinitionRouteLocator.this, route.getId(), properties))
				.bind();
		// @formatter:on
		//生成异步断言
		return factory.applyAsync(config);
	}

json复制代码private List<GatewayFilter> getFilters(RouteDefinition routeDefinition) {
		List<GatewayFilter> filters = new ArrayList<>();
		//添加默认过滤器
		if (!this.gatewayProperties.getDefaultFilters().isEmpty()) {
			filters.addAll(loadGatewayFilters(DEFAULT_FILTERS,
					new ArrayList<>(this.gatewayProperties.getDefaultFilters())));
		}
		//添加配置的过滤器
		if (!routeDefinition.getFilters().isEmpty()) {
			filters.addAll(loadGatewayFilters(routeDefinition.getId(),
					new ArrayList<>(routeDefinition.getFilters())));
		}
		//排序
		AnnotationAwareOrderComparator.sort(filters);
		return filters;
	}

List<GatewayFilter> loadGatewayFilters(String id,
			List<FilterDefinition> filterDefinitions) {
		ArrayList<GatewayFilter> ordered = new ArrayList<>(filterDefinitions.size());
		for (int i = 0; i < filterDefinitions.size(); i++) {
			FilterDefinition definition = filterDefinitions.get(i);
			//获取FilterDefinition对应的GatewayFilterFactory
			GatewayFilterFactory factory = this.gatewayFilterFactories
					.get(definition.getName());
			if (factory == null) {
				throw new IllegalArgumentException(
						"Unable to find GatewayFilterFactory with name "
								+ definition.getName());
			}
			if (logger.isDebugEnabled()) {
				logger.debug("RouteDefinition " + id + " applying filter "
						+ definition.getArgs() + " to " + definition.getName());
			}
			//生成配置类，与Predicate类似
			// @formatter:off
			Object configuration = this.configurationService.with(factory)
					.name(definition.getName())
					.properties(definition.getArgs())
					.eventFunction((bound, properties) -> new FilterArgsEvent(
							// TODO: why explicit cast needed or java compile fails
							RouteDefinitionRouteLocator.this, id, (Map<String, Object>) properties))
					.bind();
			// @formatter:on

			// some filters require routeId
			// TODO: is there a better place to apply this?
			if (configuration instanceof HasRouteId) {
				HasRouteId hasRouteId = (HasRouteId) configuration;
				hasRouteId.setRouteId(id);
			}
			//生成GatewayFilter
			GatewayFilter gatewayFilter = factory.apply(configuration);
			if (gatewayFilter instanceof Ordered) {
				ordered.add(gatewayFilter);
			}
			else {
				ordered.add(new OrderedGatewayFilter(gatewayFilter, i + 1));
			}
		}

		return ordered;
	}

bash复制代码1. 学会基本调试技能

bash复制代码git clone https://github.com/lxchuan12/nodejs-debugging.git
cd nodejs-debugging
# npm i -g yarn
yarn install

js复制代码devtool: 'source-map',

js复制代码// vue-cli 3+
module.exports = {
  configureWebpack: {
    devtool: 'source-map'
  }
}

bash复制代码git clone https://github.com/lxchuan12/nodejs-debugging.git
cd nodejs-debugging
# npm i -g yarn
yarn install

arduino复制代码$ docker run -p 6379:6379 --name redis -d redis:6.2.3-alpine

sql复制代码$ telnet 127.0.0.1 6379
Trying 127.0.0.1...
Connected to localhost.
Escape character is '^]'.
monitor
+OK

python复制代码# myapp.py
from celery import Celery

app = Celery(
    'myapp',
    broker='redis://localhost:6379/0',
    result_backend='redis://localhost:6379/0'
)

@app.task
def add(x, y):
    print("add", x, y)
    return x + y

if __name__ == '__main__':
    app.start()

ruby复制代码$ python myapp.py worker -l DEBUG

sql复制代码$ celery -A myapp worker -l DEBUG
 
 -------------- celery@bogon v5.1.2 (sun-harmonics)
--- ***** ----- 
-- ******* ---- macOS-10.16-x86_64-i386-64bit 2021-09-08 20:33:45
- *** --- * --- 
- ** ---------- [config]
- ** ---------- .> app:         myapp:0x7f855079e730
- ** ---------- .> transport:   redis://localhost:6379/0
- ** ---------- .> results:     disabled://
- *** --- * --- .> concurrency: 12 (prefork)
-- ******* ---- .> task events: OFF (enable -E to monitor tasks in this worker)
--- ***** ----- 
 -------------- [queues]
                .> celery           exchange=celery(direct) key=celery
                

[tasks]
  . myapp.add

[2021-09-08 20:33:46,220: INFO/MainProcess] Connected to redis://localhost:6379/0
[2021-09-08 20:33:46,234: INFO/MainProcess] mingle: searching for neighbors
[2021-09-08 20:33:47,279: INFO/MainProcess] mingle: all alone
[2021-09-08 20:33:47,315: INFO/MainProcess] celery@bogon ready.

python复制代码$ python
>>> from myapp import add
>>> task = add.delay(16,16)
>>> task
<AsyncResult: 5aabfc0b-04b5-4a51-86b0-6a7263e2ef3b>
>>> task.get()
32

css复制代码[2021-11-11 20:13:48,040: INFO/MainProcess] Task myapp.add[5aabfc0b-04b5-4a51-86b0-6a7263e2ef3b] received
[2021-11-11 20:13:48,040: DEBUG/MainProcess] TaskPool: Apply <function fast_trace_task at 0x7fda086baa60> (args:('myapp.add', '5aabfc0b-04b5-4a51-86b0-6a7263e2ef3b', {'lang': 'py', 'task': 'myapp.add', 'id': '5aabfc0b-04b5-4a51-86b0-6a7263e2ef3b', 'shadow': None, 'eta': None, 'expires': None, 'group': None, 'group_index': None, 'retries': 0, 'timelimit': [None, None], 'root_id': '5aabfc0b-04b5-4a51-86b0-6a7263e2ef3b', 'parent_id': None, 'argsrepr': '(16, 16)', 'kwargsrepr': '{}', 'origin': 'gen63119@localhost', 'ignore_result': False, 'reply_to': '97a3e117-c8cf-3d4c-97c0-c0a76aaf9a16', 'correlation_id': '5aabfc0b-04b5-4a51-86b0-6a7263e2ef3b', 'hostname': 'celery@localhost', 'delivery_info': {'exchange': '', 'routing_key': 'celery', 'priority': 0, 'redelivered': None}, 'args': [16, 16], 'kwargs': {}}, b'[[16, 16], {}, {"callbacks": null, "errbacks": null, "chain": null, "chord": null}]', 'application/json', 'utf-8') kwargs:{})
[2021-11-11 20:13:49,059: INFO/ForkPoolWorker-8] Task myapp.add[5aabfc0b-04b5-4a51-86b0-6a7263e2ef3b] succeeded in 1.0166977809999995s: 32

swift复制代码+1636632828.304020 [0 172.16.0.117:51127] "SUBSCRIBE" "celery-task-meta-5aabfc0b-04b5-4a51-86b0-6a7263e2ef3b"
+1636632828.304447 [0 172.16.0.117:51129] "PING"
+1636632828.305448 [0 172.16.0.117:51129] "LPUSH" "celery" "{\"body\": \"W1sxNiwgMTZdLCB7fSwgeyJjYWxsYmFja3MiOiBudWxsLCAiZXJyYmFja3MiOiBudWxsLCAiY2hhaW4iOiBudWxsLCAiY2hvcmQiOiBudWxsfV0=\", \"content-encoding\": \"utf-8\", \"content-type\": \"application/json\", \"headers\": {\"lang\": \"py\", \"task\": \"myapp.add\", \"id\": \"5aabfc0b-04b5-4a51-86b0-6a7263e2ef3b\", \"shadow\": null, \"eta\": null, \"expires\": null, \"group\": null, \"group_index\": null, \"retries\": 0, \"timelimit\": [null, null], \"root_id\": \"5aabfc0b-04b5-4a51-86b0-6a7263e2ef3b\", \"parent_id\": null, \"argsrepr\": \"(16, 16)\", \"kwargsrepr\": \"{}\", \"origin\": \"gen63119@localhost\", \"ignore_result\": false}, \"properties\": {\"correlation_id\": \"5aabfc0b-04b5-4a51-86b0-6a7263e2ef3b\", \"reply_to\": \"97a3e117-c8cf-3d4c-97c0-c0a76aaf9a16\", \"delivery_mode\": 2, \"delivery_info\": {\"exchange\": \"\", \"routing_key\": \"celery\"}, \"priority\": 0, \"body_encoding\": \"base64\", \"delivery_tag\": \"20dbd584-b669-4ef0-8a3b-41d19b354690\"}}"
+1636632828.307040 [0 172.16.0.117:52014] "MULTI"
+1636632828.307075 [0 172.16.0.117:52014] "ZADD" "unacked_index" "1636632828.038743" "20dbd584-b669-4ef0-8a3b-41d19b354690"
+1636632828.307088 [0 172.16.0.117:52014] "HSET" "unacked" "20dbd584-b669-4ef0-8a3b-41d19b354690" "[{\"body\": \"W1sxNiwgMTZdLCB7fSwgeyJjYWxsYmFja3MiOiBudWxsLCAiZXJyYmFja3MiOiBudWxsLCAiY2hhaW4iOiBudWxsLCAiY2hvcmQiOiBudWxsfV0=\", \"content-encoding\": \"utf-8\", \"content-type\": \"application/json\", \"headers\": {\"lang\": \"py\", \"task\": \"myapp.add\", \"id\": \"5aabfc0b-04b5-4a51-86b0-6a7263e2ef3b\", \"shadow\": null, \"eta\": null, \"expires\": null, \"group\": null, \"group_index\": null, \"retries\": 0, \"timelimit\": [null, null], \"root_id\": \"5aabfc0b-04b5-4a51-86b0-6a7263e2ef3b\", \"parent_id\": null, \"argsrepr\": \"(16, 16)\", \"kwargsrepr\": \"{}\", \"origin\": \"gen63119@localhost\", \"ignore_result\": false}, \"properties\": {\"correlation_id\": \"5aabfc0b-04b5-4a51-86b0-6a7263e2ef3b\", \"reply_to\": \"97a3e117-c8cf-3d4c-97c0-c0a76aaf9a16\", \"delivery_mode\": 2, \"delivery_info\": {\"exchange\": \"\", \"routing_key\": \"celery\"}, \"priority\": 0, \"body_encoding\": \"base64\", \"delivery_tag\": \"20dbd584-b669-4ef0-8a3b-41d19b354690\"}}, \"\", \"celery\"]"
...

python复制代码# ch23-celery/celery-5.0.5/celery/__main__.py
def main():
    """Entrypoint to the ``celery`` umbrella command."""
    """celery命令入口"""
    ...
    # 具体执行的main函数
    from celery.bin.celery import main as _main
    sys.exit(_main())

python复制代码# ch23-celery/celery-5.0.5/celery/bin/celery.py
def celery(ctx, app, broker, result_backend, loader, config, workdir,
           no_color, quiet, version):
    """Celery command entrypoint."""
    ...
    ctx.obj = CLIContext(app=app, no_color=no_color, workdir=workdir,
                         quiet=quiet)
    # worker/beat/events三个主要子命令参数
    # User options
    worker.params.extend(ctx.obj.app.user_options.get('worker', []))
    beat.params.extend(ctx.obj.app.user_options.get('beat', []))
    events.params.extend(ctx.obj.app.user_options.get('events', []))

def main() -> int:
    """Start celery umbrella command.

    This function is the main entrypoint for the CLI.

    :return: The exit code of the CLI.
    """
    return celery(auto_envvar_prefix="CELERY")

ini复制代码# ch23-celery/celery-5.0.5/celery/bin/worker.py
def worker(ctx, hostname=None, pool_cls=None, app=None, uid=None, gid=None,
           loglevel=None, logfile=None, pidfile=None, statedb=None,
           **kwargs):
    # 创建和启动worker
    worker = app.Worker(
        hostname=hostname, pool_cls=pool_cls, loglevel=loglevel,
        logfile=logfile,  # node format handled by celery.app.log.setup
        pidfile=node_format(pidfile, hostname),
        statedb=node_format(statedb, hostname),
        no_color=ctx.obj.no_color,
        quiet=ctx.obj.quiet,
        **kwargs)
    worker.start()

ruby复制代码# ch23-celery/celery-5.0.5/celery/app/base.py
def Worker(self):
    # 创建worker
    return self.subclass_with_self('celery.apps.worker:Worker')

lua复制代码                                 +----------+
                             +--->app.celery|
                             |   +----------+
+---------+   +----------+   |
|main.main+--->bin.celery+---+
+---------+   +----------+   |
                             |   +----------+   +-----------+
                             +--->bin.worker+--->apps.worker|
                                 +----------+   +-----------+

python复制代码# ch23-celery/celery-5.0.5/celery/app/base.py
task_cls = 'celery.app.task:Task'

def task(self, *args, **opts):
    """Decorator to create a task class out of any callable.
    """
    def inner_create_task_cls(shared=True, filter=None, lazy=True, **opts):
        
        def _create_task_cls(fun):
            
            ret = PromiseProxy(self._task_from_fun, (fun,), opts,
                                       __doc__=fun.__doc__)
            return ret

        return _create_task_cls
    return inner_create_task_cls(**opts)

kotlin复制代码def _task_from_fun(self, fun, name=None, base=None, bind=False, **options):
    base = base or self.Task
    task = type(fun.__name__, (base,), dict({
                'app': self,
                'name': name,
                'run': run,
                '_decorated': True,
                '__doc__': fun.__doc__,
                '__module__': fun.__module__,
                '__annotations__': fun.__annotations__,
                '__header__': staticmethod(head_from_fun(fun, bound=bind)),
                '__wrapped__': run}, **options))
    add_autoretry_behaviour(task, **options)
    # 增加task
    self._tasks[task.name] = task
    task.bind(self)  # connects task to this app
    add_autoretry_behaviour(task, **options)
    return task

ruby复制代码# ch23-celery/celery-5.0.5/celery/app/task.py
def delay(self, *args, **kwargs):
    ...
    return app.send_task(
                self.name, args, kwargs, task_id=task_id, producer=producer,
                link=link, link_error=link_error, result_cls=self.AsyncResult,
                shadow=shadow, task_type=self,
                **options
            )

ini复制代码class Celery:
    
    # 协议类
    amqp_cls = 'celery.app.amqp:AMQP'
    backend_cls = None
    # 事件类
    events_cls = 'celery.app.events:Events'
    loader_cls = None
    log_cls = 'celery.app.log:Logging'
    # 控制类
    control_cls = 'celery.app.control:Control'
    # 任务类
    task_cls = 'celery.app.task:Task'
    # 任务注册中心
    registry_cls = 'celery.app.registry:TaskRegistry'
    ...
    
    def __init__(self, main=None, loader=None, backend=None,
                 amqp=None, events=None, log=None, control=None,
                 set_as_current=True, tasks=None, broker=None, include=None,
                 changes=None, config_source=None, fixups=None, task_cls=None,
                 autofinalize=True, namespace=None, strict_typing=True,
                 **kwargs):
        # 启动步骤
        self.steps = defaultdict(set)
        # 待执行的task
        self._pending = deque()
        # 所有任务
        self._tasks = self.registry_cls(self._tasks or {})
        ...
        self.__autoset('broker_url', broker)
        self.__autoset('result_backend', backend)
        ...
        self.on_init()
        _register_app(self)

ruby复制代码def task(self, *args, **opts):
    ...
    # 增加task
    self._tasks[task.name] = task
    task.bind(self)  # connects task to this app
    add_autoretry_behaviour(task, **options)
    ...

python复制代码def _connection(self, url, userid=None, password=None,
                virtual_host=None, port=None, ssl=None,
                connect_timeout=None, transport=None,
                transport_options=None, heartbeat=None,
                login_method=None, failover_strategy=None, **kwargs):
    conf = self.conf
    return self.amqp.Connection(
        url,
        userid or conf.broker_user,
        password or conf.broker_password,
        virtual_host or conf.broker_vhost,
        port or conf.broker_port,
        transport=transport or conf.broker_transport,
        ssl=self.either('broker_use_ssl', ssl),
        heartbeat=heartbeat,
        login_method=login_method or conf.broker_login_method,
        failover_strategy=(
            failover_strategy or conf.broker_failover_strategy
        ),
        transport_options=dict(
            conf.broker_transport_options, **transport_options or {}
        ),
        connect_timeout=self.either(
            'broker_connection_timeout', connect_timeout
        ),
    )
broker_connection = connection

@cached_property
def amqp(self):
    """AMQP related functionality: :class:`~@amqp`."""
    return instantiate(self.amqp_cls, app=self)

python复制代码from kombu import Connection, Consumer, Exchange, Producer, Queue, pools

class AMQP:
    """App AMQP API: app.amqp."""

    Connection = Connection

ini复制代码def Queues(self, queues, create_missing=None,
           autoexchange=None, max_priority=None):
    ...
    return self.Queues(
            queues, self.default_exchange, create_missing,
            autoexchange, max_priority, default_routing_key,
        )
        
def TaskConsumer(self, channel, queues=None, accept=None, **kw):
    ...
    return self.Consumer(
        channel, accept=accept,
        queues=queues or list(self.queues.consume_from.values()),
        **kw
    )

def _create_task_sender(self):
    ...
    producer.publish(
                body,
                exchange=exchange,
                routing_key=routing_key,
                serializer=serializer or default_serializer,
                compression=compression or default_compressor,
                retry=retry, retry_policy=_rp,
                delivery_mode=delivery_mode, declare=declare,
                headers=headers2,
                **properties
            )
    ...

python复制代码class WorkController:
    # 内部类
    class Blueprint(bootsteps.Blueprint):
        """Worker bootstep blueprint."""

        name = 'Worker'
        default_steps = {
            'celery.worker.components:Hub',
            'celery.worker.components:Pool',
            'celery.worker.components:Beat',
            'celery.worker.components:Timer',
            'celery.worker.components:StateDB',
            'celery.worker.components:Consumer',
            'celery.worker.autoscale:WorkerComponent',
        }
    
    def __init__(self, app=None, hostname=None, **kwargs):
        self.blueprint = self.Blueprint(
            steps=self.app.steps['worker'],
            on_start=self.on_start,
            on_close=self.on_close,
            on_stopped=self.on_stopped,
        )
        self.blueprint.apply(self, **kwargs)

python复制代码def start(self):
    try:
        # 启动worker
        self.blueprint.start(self)
    except WorkerTerminate:
        self.terminate()
    except Exception as exc:
        logger.critical('Unrecoverable error: %r', exc, exc_info=True)
        self.stop(exitcode=EX_FAILURE)
    except SystemExit as exc:
        self.stop(exitcode=exc.code)
    except KeyboardInterrupt:
        self.stop(exitcode=EX_FAILURE)

python复制代码class StepType(type):
    """Meta-class for steps."""

    name = None
    requires = None

class Step(metaclass=StepType):
    ...
    
    def instantiate(self, name, *args, **kwargs):
        return symbol_by_name(name)(*args, **kwargs)
    
    def include_if(self, parent):
        return self.enabled
        
    def _should_include(self, parent):
        if self.include_if(parent):
            return True, self.create(parent)
        return False, None

    def create(self, parent):
        """Create the step."""

python复制代码class ConsumerStep(StartStopStep):
    """Bootstep that starts a message consumer."""

    requires = ('celery.worker.consumer:Connection',)
    consumers = None

    def start(self, c):
        channel = c.connection.channel()
        self.consumers = self.get_consumers(channel)
        for consumer in self.consumers or []:
            consumer.consume()

class Connection(bootsteps.StartStopStep):
    """Service managing the consumer broker connection."""

    def __init__(self, c, **kwargs):
        c.connection = None
        super().__init__(c, **kwargs)

    def start(self, c):
        c.connection = c.connect()
        info('Connected to %s', c.connection.as_uri())