java复制代码@Scheduled("*************************")
public void pushSpikeInfoIntoRedis(){
       //查询未开始的秒杀活动信息
        List<RedisSpike> spikeList = spikeMapper.selectSpikeOfNotStart();
      
        //查询秒杀商品信息
        List<RedisSpikeSku> spikeSkuList = spikeSkuMapper.selectSkuList();
        //根据活动分组
       Map<Long, List<RedisSpikeSku>> groupSpikeSkuList=
       spikeSkuList.stream().collect(Collectors.groupingBy(RedisSpikeSku::getSpikeId))

        //秒杀活动存储到redis
        for (RedisSpike spike : spikeList) {
            redisTemplate.opsForHash()
           .put("hosjoy-spike-test:spike-info", spikeId,spike );
           
         }
        //秒杀商品存储到redis
       for(Map.Entry<Long, List<RedisSpikeSku>>entry: groupSpikeSkuList.entrySet()){
           Long spikeId= entry.getKey;
           List<RedisSpikeSku >skuList = entry.getValue();
           for(RedisSpikeSku sku: skuList){
            redisTemplate.opsForHash()
           .put("hosjoy-spike-test:spike-sku:"+spikeId, sku.getId, sku );                 
           }      
        }       
    
}

java复制代码        //库存信息
      for(RedisSpikeSku sku: spikeSkuList){
          Integer inventory = sku.getInventory();
          List<Long> skuIds = new ArrayList<>(inventory);
          for(int i=0; i<inventory; i++){
              skuIds.add(sku.getId());
          }          
          redisTemplate.opsForList().leftPushAll("hosjoy-spike-test:spike-sku-inventory:" + spikeId + spikeSkuId, skuIds);
      }

java复制代码    //删除已过期活动
    List<Long> expireSpikes = new ArrayList<>();
    List<RedisSpike> spikes = redisTemplate.opsForHash().values("hosjoy-spike-test:spike-info");
    LocalDateTime now = LocalDateTime.now();
    spikes.forEach(s->{
        LocalDateTime endTime = s.getEndTime();
        if(now.after(endTime)){
            expireSpikes.add(s.getId());
            redisTemplate.opsForHash().delete("hosjoy-spike-test:spike-info",s.getId());
        }
     });

    //删除已过期sku
    expireSpikes.forEach(spikeId->{
        redisTemplate.opsForHash().delete("hosjoy-spike-test:spike-sku:"+spikeId);
     });

java复制代码public SpikeListResponse spikeList(Long userId) {    
    //获取所有秒杀活动
    List<RedisSpike> spikes = redisTemplate.opsForHash().values("hosjoy-spike-test:spike-     info");
    //该用户可见的活动
    List<RedisSpike> visibleSpikes= new ArrayList<>();
    
    //业务处理，过滤当前用户不可见的活动
    ////根据具体业务自由实现
    removeSomeSpike(spikes, userId, visibleSpikes);
    
    //距离当前时间最近的活动
    RedisSpikeSku currentSpike = getCurrentSpike(visibleSpikes);
    
    //获取该活动商品列表
    List<RedisSpikeSku> spikeSkuList = redisTemplate.opsForHash()
           .values("hosjoy-spike-test:spike-sku:"+currentSpike.getId() );
    
    //业务校验、处理(例如去除当前用户不可见的商品)
    //根据具体业务自由实现
    removeSomeSku(spikeSkuList, userId);
    
    //数据封装
    SpikeListResponse spikeList =new SpikeListResponse();
    spikeList.setSpikes(visibleSpikes); //活动列表
    spikeList.setCurrentSpike(currentSpike); //当前活动
    spikeList.setCurrentSpikeSkuList(spikeSkuList); //商品列表
    return spikeList;
}

java复制代码public SpikeSkuListResponse spikeSkuListResponse(Long spikeId, Long userId) {    
    //获取活动信息
    RedisSpike spike = redisTemplate.opsForHash()
           .get("hosjoy-spike-test:spike-info", spikeId);
    
    //获取该活动商品列表
    List<RedisSpikeSku> spikeSkuList = redisTemplate.opsForHash()
           .values("hosjoy-spike-test:spike-sku:"+spikeId );
    
     //业务校验、处理(例如去除当前用户不可见的商品)
    //根据具体业务自由实现
    removeSomeSku(spikeSkuList, userId);
    
    //数据封装
    SpikeSkuListResponse response =new SpikeSkuListResponse();
    response.spikeInfo(spike); //活动信息
    response.setSpikeSkuList(spikeSkuList); //商品列表
    return spikeList;
}

java复制代码public SpikeSkuResponse spikeSkuDetail(Long spikeId, Long skuId) {
   //获取商品信息
   RedisSpikeSku sku = redisTemplate.opsForHash().get("hosjoy-spike-test:spike-           sku:"+spikeId, skuId);
    
    //获取活动信息
    RedisSpike spike = redisTemplate.opsForHash()
           .get("hosjoy-spike-test:spike-info", spikeId);
    
    //其它业务处理
    .............................
    .............................
    
    //数据封装
    SpikeSkuResponse response =new SpikeSkuResponse();
    response.set...........
    response.set...........
    return response;
}

java复制代码//判断是否已排队
//防止脚本刷单、重复提交等，
Long currentUserQueue = redisTemplate.opsForHash().get("hosjoy-spike-test:spike-user-queue-count:"+ spikeId+spikeSkuId, userId);
if(currentUserQueue > 1){
    throw new ClientException("您已提交了订单，请误重复提交");
}

java复制代码Long failCount= redisTemplate.opsForHash().get("hosjoy-spike-test:spike-black-user:"+         spikeId+spikeSkuId, userId);

if(failCount > SpikeConstants.MAX_FAIL_TIMES){
    throw new ClientException("您不符合购买条件");
}

java复制代码//当前用户提交次数
Long currentUserQueue = redisTemplate.opsForHash().increment("hosjoy-spike-test:spike-user-queue-count:"+ spikeId+spikeSkuId, userId, 1);
if(currentUserQueue > 1){
    throw new ClientException("您已提交了订单，请勿重复提交");
}

sql复制代码select inventory from sku;
update sku set inventory =#{inventory} where id=?  and inventory= ?

java复制代码//预扣减库存
String  skuId = redisTemplate..opsForList().rightPop("hosjoy-spike-test:spike-sku-inventory:" + spikeId + spikeSkuId);
//pop出来空，说明已售罄
//此处不需要加锁判断，不会产生并发问题
if(skuId == null){
    //删除排队信息
   redisTemplate.opsForHash().delete("hosjoy-spike-test:spike-user-queue-count:"+        spikeId+spikeSkuId, userId);   
    throw new ClientException("商品已售空");
}

java复制代码//下单状态存入到Redis中 状态为排队中
OrderQueueState orderQueueState = new OrderQueueState(......下单状态封装........);
redisTemplate.opsForHash().put("hosjoy-spike-test:spike-order-user-queue-state:"+ spikeId+spikeSkuId, userId, orderQueueState);

//更新秒杀商品库存数量
 redisTemplate.opsForHash()
           .increment("hosjoy-spike-test:spike-sku:"+spikeId, skuId, -1 );

//发送排队信息到RabbitMq
sendRabbit(queueState);

java复制代码@Transactional
@StreamListener("USER-SPIKE-ORDER-QUEUE")
public void userSpikeOrderQueueInput(Message message) {
        
    //调用订单创建逻辑，入库，生成订单id
    Long orderId = appOrderService.submit();
    //订单创建成功，更新下单状态为订单提交成功
    if(orderId != null){        
        OrderQueueState orderQueueState = new OrderQueueState();
        orderQueueState.setState("订单创建成功")
        orderQueueState.setOrderId(orderId);
        
        redisTemplate.opsForHash().put("hosjoy-spike-test:spike-order-user-queue-state:"+         spikeId+spikeSkuId, userId, orderQueueState);
        //发送延迟消息，超时未支付自动关闭订单
        sendMQDelayMessage();
    }
    //可能因为各种业务校验不通过而导致提交订单失败
    else{
        //回滚库存队列，否则会出现少卖问题
        redisTemplate.opsForList.leftPush("hosjoy-spike-test:spike-sku-inventory:" + spikeId + spikeSkuId, spikeSkuId);
        //商品库存数量+1
        redisTemplate.opsForHash()
           .increment("hosjoy-spike-test:spike-sku:"+spikeId, spikeSkuId, 1 );
       
        //下单状态更新为订单提交失败
        OrderQueueState orderQueueState = new OrderQueueState();
        orderQueueState.setState("订单创建失败")
        redisTemplate.opsForHash().put("hosjoy-spike-test:spike-order-user-queue-state:"+         spikeId+spikeSkuId, userId, orderQueueState); 
        //删除排队信息
        redisTemplate.opsForHash().delete("hosjoy-spike-test:spike-user-queue-count:"+             spikeId+spikeSkuId, userId);
        //用户多次提交订单失败，有可能是恶意刷单，也可能是该用户不符合购买条件导致提交订单失败（实际业务校验          不通过）。 
        //可以将该用户加入黑名单。
        redisTemplate.opsForHash().increment("hosjoy-spike-test:spike-black-user:"+         spikeId+spikeSkuId, userId, 1)
    }    
}

java复制代码@Transactional
@StreamListener(OrderSink.UNPAID_AUTO_CLOSE_INPUT)
public void unpaidAutoCloseInput(Long orderId) {
    log.info("[RABBITMQ][ORDER] 收到订单订单自动关闭消息, 订单id为: {}", orderId);
    this.closeOrder();
    
    //看看是否确实没有支付
      OrderQueueState orderQueueState = redisTemplate.opsForHash().get("hosjoy-spike-test:spike-order-user-queue-state:"+spikeId+spikeSkuId, userId);
    //仍然有下单排队状态缓存，说明没有完成支付
    if(orderQueueState!=null){
     //回滚库存
    redisTemplate.opsForList.leftPush("hosjoy-spike-test:spike-sku-inventory:" + spikeId +     spikeSkuId, spikeSkuId);
     //删除排队信息
    redisTemplate.opsForHas h().delete("hosjoy-spike-test:spike-user-queue-count:"+            spikeId+spikeSkuId, userId); 
      //删除下单排队状态
       redisTemplate.opsForHash().delete("hosjoy-spike-test:spike-order-user-queue-              state:"+ spikeId+spikeSkuId, userId);   
    }
    
}

java复制代码//删除排队信息
 redisTemplate.opsForHash().delete("hosjoy-spike-test:spike-user-queue-count:"+        spikeId+spikeSkuId);

java复制代码//删除下单排队状态缓存
//如果不删除，在收到超时未支付消息时，会认为该订单未支付
 redisTemplate.opsForHash().delete("hosjoy-spike-test:spike-order-user-queue-state:"+        spikeId+spikeSkuId, userId);

java复制代码@GetMapping(value = "/status")
public Integer queryStatus(){
       
    OrderQueueState orderQueueState=
        redisTemplate.opsForHash().get("hosjoy-spike-test:spike-order-user-queue-state:"+         spikeId+spikeSkuId, userId);
    
    if(orderQueueState!=null){
        orderQueueState.getState();
    }
    //查询不到下单状态缓存
    //可能原因：1.未下单 2：创建订单业务逻辑出错 3: 超时未支付 4：已支付
    throw new ClientException("ERROR");
}

ini复制代码limit_conn_zone $binary_remote_address zone=addr:10m;

java复制代码//用于加密和解密的密钥
private final String cipher="hosjoy-spike-md5-cipher&73@(**$d--=,./;~·2··%##4";

//返回url可变部分
public String  getPathVirableMd5(long skuId) {
    if(LocalDateTime.now().before(spikeStartTime) ){
        throw new ClientException("还没到秒杀时间")
    }
    String base=skuId+"/"+cipher;
    return DigestUtils.md5DigestAsHex(base.getBytes());
}

java复制代码@RequestMapping(value = "/submit/{skuId}/{md5}")
public void  submit(@PathVariable("skuId") Long skuId, @PathVariable("md5") String md5){
    return spikeService.submitSpikeOrder(skuId, md5);  
}

public void submitSpikeOrder(long skuId,  String md5){
      if(md5==null||!md5.equals(getPathVirableMd5(skuId))){
         //md5校验错误
         throw new RuntimeException("非法请求");
      }
      //校验通过，执行下单
      this. doSubmitSpikeOrder(skuId);
    }
}

java复制代码@Autowired
StringRedisTemplate redisTemplate;

/**
     * 模拟商品减库存操作
     * @param productCode
     */
@PutMapping("reduce/{product-code}")
public void reduce(@PathVariable(value = "product-code") String productCode) {
    String lockKey = "lock:" + productCode;
    try {
        Boolean isSuccess = redisTemplate.opsForValue().setIfAbsent(lockKey, productCode, 10, TimeUnit.MILLISECONDS);

        if (isSuccess) {
            // 获取锁成功 执行减库存操作
            Integer count = Integer.parseInt(redisTemplate.opsForValue().get(productCode));
            if (count > 0) {
                redisTemplate.opsForValue().increment(productCode);
            }
        }
    } finally {
        // 释放锁
        redisTemplate.delete(lockKey);
    }
}

xml复制代码<dependencies>
  <dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
  </dependency>
<dependency>
  <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.14.0</version>
  </dependency>
</dependencies>

yaml复制代码
       server:
         port: 9000
       spring:
         redis:
           host: 47.102.218.26
           password: root
           port: 6379
           cluster:
             failed-attempts: 3
             master-connection-pool-size: 64
             nodes: ''
             read-mode: SLAVE
             retry-attempts: 3
             retry-interval: 1500
             scan-interval: 1000
             slave-connection-pool-size: 64
           pool:
             conn-timeout: 3000
             max-active: 8
             max-idle: 16
             max-wait: 3000
             min-idle: 8
             size: 10
             so-timeout: 3000
           sentinel:
             fail-max: ''
             master: business-master
             master-onlyWrite: true
             nodes: ''
           single:
             address: 192.168.60.23:6379

scss复制代码   @Data
   @ConfigurationProperties(prefix = "redisson")
   public class RedissonProperties {
   
       private String address;
   
       private int database = 0;
   
       private String password;
   
       private int timeout;
   
       /**
        * 池配置
        */
       private RedisPoolProperties pool;
   
       /**
        * 单机
        */
       private RedisSingleProperties single;
   
       /**
        * 哨兵
        */
       private RedissonSentinelProperties sentinel;
   
       /**
        * 主从
        */
       private RedissonMasterSlaveProperties masterSlave;
   
       /**
        * 集群
        */
       private RedissonClusterProperties cluster;
   
   
   }
   

   /**
    * @author xiangjin.kong
    * @date 2021/3/25 10:32
    */
   @Configuration
   @ConditionalOnClass({Redisson.class, RedisOperations.class})
   @EnableAutoConfiguration
   @EnableConfigurationProperties(RedissonProperties.class)
   public class RedissonAutoConfiguration {
   
       @Autowired
       RedissonProperties redisProperties;
       /**
        * 单机
        * @return
        */
       /**
        * 单机模式 redisson 客户端
        */
   
       @Bean
       @ConditionalOnProperty(name = "spring.redis.mode", havingValue = "single")
       RedissonClient redissonSingle() {
           Config config = new Config();
           String node = redisProperties.getSingle().getAddress();
           node = node.startsWith("redis://") ? node : "redis://" + node;
           SingleServerConfig serverConfig = config.useSingleServer()
                   .setAddress(node)
                   .setTimeout(redisProperties.getPool().getConnTimeout())
                   .setConnectionPoolSize(redisProperties.getPool().getSize())
                   .setConnectionMinimumIdleSize(redisProperties.getPool().getMinIdle());
           if (StringUtils.isNotBlank(redisProperties.getPassword())) {
               serverConfig.setPassword(redisProperties.getPassword());
           }
           return Redisson.create(config);
       }
   
   
       /**
        * 集群模式的 redisson 客户端
        *
        * @return
        */
       @Bean
       @ConditionalOnProperty(name = "spring.redis.mode", havingValue = "cluster")
       RedissonClient redissonCluster() {
           System.out.println("cluster redisProperties:" + redisProperties.getCluster());
   
           Config config = new Config();
           String[] nodes = redisProperties.getCluster().getNodes().split(",");
           List<String> newNodes = new ArrayList(nodes.length);
           Arrays.stream(nodes).forEach((index) -> newNodes.add(
                   index.startsWith("redis://") ? index : "redis://" + index));
   
           ClusterServersConfig serverConfig = config.useClusterServers()
                   .addNodeAddress(newNodes.toArray(new String[0]))
                   .setScanInterval(
                           redisProperties.getCluster().getScanInterval())
                   .setIdleConnectionTimeout(
                           redisProperties.getPool().getSoTimeout())
                   .setConnectTimeout(
                           redisProperties.getPool().getConnTimeout())
                   .setRetryAttempts(
                           redisProperties.getCluster().getRetryAttempts())
                   .setRetryInterval(
                           redisProperties.getCluster().getRetryInterval())
                   .setMasterConnectionPoolSize(redisProperties.getCluster()
                           .getMasterConnectionPoolSize())
                   .setSlaveConnectionPoolSize(redisProperties.getCluster()
                           .getSlaveConnectionPoolSize())
                   .setTimeout(redisProperties.getTimeout());
           if (StringUtils.isNotBlank(redisProperties.getPassword())) {
               serverConfig.setPassword(redisProperties.getPassword());
           }
           return Redisson.create(config);
       }
   
       /**
        * 哨兵模式 redisson 客户端
        * @return
        */
   
       @Bean
       @ConditionalOnProperty(name = "spring.redis.mode", havingValue = "sentinel")
       RedissonClient redissonSentinel() {
           System.out.println("sentinel redisProperties:" + redisProperties.getSentinel());
           Config config = new Config();
           String[] nodes = redisProperties.getSentinel().getNodes().split(",");
           List<String> newNodes = new ArrayList(nodes.length);
           Arrays.stream(nodes).forEach((index) -> newNodes.add(
                   index.startsWith("redis://") ? index : "redis://" + index));
   
           SentinelServersConfig serverConfig = config.useSentinelServers()
                   .addSentinelAddress(newNodes.toArray(new String[0]))
                   .setMasterName(redisProperties.getSentinel().getMaster())
                   .setReadMode(ReadMode.SLAVE)
                   .setTimeout(redisProperties.getTimeout())
                   .setMasterConnectionPoolSize(redisProperties.getPool().getSize())
                   .setSlaveConnectionPoolSize(redisProperties.getPool().getSize());
   
           if (StringUtils.isNotBlank(redisProperties.getPassword())) {
               serverConfig.setPassword(redisProperties.getPassword());
           }
   
           return Redisson.create(config);
       }
   }
   
   @Data
   public class RedisPoolProperties {
       private int maxIdle;
   
       private int minIdle;
   
       private int maxActive;
   
       private int maxWait;
   
       private int connTimeout;
   
       private int soTimeout;
   
       /**
        * 池大小
        */
       private  int size;
   
   }
   
   
   @Data
   public class RedisSingleProperties {
   
       private String address;
   }
   
   
   @Data
   public class RedissonClusterProperties {
   
       /**
        * 集群状态扫描间隔时间，单位是毫秒
        */
       private int scanInterval;
   
       /**
        * 集群节点
        */
       private String nodes;
   
       /**
        * 默认值： SLAVE（只在从服务节点里读取）设置读取操作选择节点的模式。 可用值为： SLAVE - 只在从服务节点里读取。
        * MASTER - 只在主服务节点里读取。 MASTER_SLAVE - 在主从服务节点里都可以读取
        */
       private String readMode;
       /**
        * （从节点连接池大小） 默认值：64
        */
       private int slaveConnectionPoolSize;
       /**
        * 主节点连接池大小）默认值：64
        */
       private int masterConnectionPoolSize;
   
       /**
        * （命令失败重试次数） 默认值：3
        */
       private int retryAttempts;
   
       /**
        * 命令重试发送时间间隔，单位：毫秒 默认值：1500
        */
       private int retryInterval;
   
       /**
        * 执行失败最大次数默认值：3
        */
       private int failedAttempts;
   }
   
   
   @Data
   public class RedissonMasterSlaveProperties {
   
       private String masterAddress;
       private String slaveAddress;
   
   }
   
   
   @Data
   public class RedissonSentinelProperties {
   
       /**
        * 哨兵master 名称
        */
       private String master;
   
       /**
        * 哨兵节点
        */
       private String nodes;
   
       /**
        * 哨兵配置
        */
       private boolean masterOnlyWrite;
   
       /**
        *
        */
       private int failMax;
   
   }
 }

kotlin复制代码public class UserRepository {
    private final UserLocalDataSource userLocalDataSource;
    private final UserRemoteDataSource userRemoteDataSource;

    public UserRepository(UserLocalDataSource userLocalDataSource, UserRemoteDataSource userRemoteDataSource) {
        this.userLocalDataSource = userLocalDataSource;
        this.userRemoteDataSource = userRemoteDataSource;
    }
}

scss复制代码new UserRepository(new UserLocalDataSource(), new UserRemoveDataSource());

csharp复制代码public static UserRepository get() {
    return new UserRepository(new UserLocalDataSource(), new UserRemoveDataSource());
}

csharp复制代码@Component
public interface ApplicationComponent {
    UserRepository userRepository();
}

kotlin复制代码public class UserRepository {

    private final UserLocalDataSource userLocalDataSource;
    private final UserRemoteDataSource userRemoteDataSource;

    @Inject
    public UserRepository(UserLocalDataSource userLocalDataSource, UserRemoteDataSource userRemoteDataSource) {
        this.userLocalDataSource = userLocalDataSource;
        this.userRemoteDataSource = userRemoteDataSource;
    }
}
--------------------------------------------
public class UserLocalDataSource {
    @Inject
    public UserLocalDataSource() {
    }
}
--------------------------------------------
public class UserRemoveDataSource {
    @Inject
    public UserRemoveDataSource() {
    }
}

csharp复制代码1、实现 ApplicationComponent 接口
public final class DaggerApplicationComponent implements ApplicationComponent {
    private DaggerApplicationComponent() {
    }
    
    2、创建依赖项实例
    @Override
    public UserRepository userRepository() {
        return new UserRepository(new UserLocalDataSource(), new UserRemoteDataSource());
    }

    3、构建者模式
    public static Builder builder() {
        return new Builder();
    }

    public static ApplicationComponent create() {
        return new Builder().build();
    }

    public static final class Builder {
        private Builder() {
        }

        public ApplicationComponent build() {
            return new DaggerApplicationComponent();
        }
    }
}

ini复制代码ApplicationComponent component = DaggerApplicationComponent.create();

UserRepository userRepository = component.userRepository();

kotlin复制代码构造器注入：（X）
public class MyActivity {
    @Inject
    public MyActivity(LoginViewModel viewModel){
        ...
    }
}
--------------------------------------------
字段注入：
class MainActivity : AppCompatActivity() {
    @Inject
    lateinit var viewModel: LoginViewModel

    override fun onCreate(savedInstanceState: Bundle?) {
        DaggerApplicationComponent.create().inject001(this)
        super.onCreate(savedInstanceState)
        ...
    }
}
public class LoginViewModel {
    private final UserRepository userRepository;

    @Inject
    public LoginViewModel(UserRepository userRepository) {
        this.userRepository = userRepository;
    }
}

less复制代码@Singleton
public class UserRepository {
    ...
}
--------------------------------------------
@Component
@Singleton
public interface ApplicationComponent {
    ...
}

less复制代码@Scope
@Documented
@Retention(RUNTIME)
public @interface Singleton {}
--------------------------------------------
@Scope
@Retention(RetentionPolicy.RUNTIME)
public @interface MyCustomScope {}

csharp复制代码public final class DaggerApplicationComponent implements ApplicationComponent {
    private Provider<UserRepository> userRepositoryProvider;

    private DaggerApplicationComponent() {
        initialize();
    }

    private void initialize() {
        this.userRepositoryProvider = DoubleCheck.provider(UserRepository_Factory.create(UserLocalDataSource_Factory.create(), UserRemoteDataSource_Factory.create()));
    }

    @Override
    public UserRepository userRepository() {
        return userRepositoryProvider.get();
    }
    ...
}

less复制代码按照业务划分：
@Singleton
@LoginScope
@RegisterScope
--------------------------------------------
按声明周期划分：
@Singleton
@ActivityScope
@ModuleScope
@FeatureScope

java复制代码public class UserRemoteDataSource {
    private final LoginRetrofitService loginRetrofitService;
    @Inject
    public UserRemoteDataSource(LoginRetrofitService loginRetrofitService) {
        this.loginRetrofitService = loginRetrofitService;
    }
}
--------------------------------------------
@Module
public class NetworkModule {
    @Provides
    public LoginRetrofitService provide001(OkHttpClient client) {
        return new Retrofit.Builder()
                .baseUrl("https://example.com")
                .build()
                .create(LoginService.class);
    }
}
--------------------------------------------
@Singleton
@Component(modules = NetworkModule.class)
public interface ApplicationComponent {

    UserRepository userRepository();

    void inject001(MainActivity activity);
}

java复制代码@Subcomponent
public interface LoginComponent {
    void inject(LoginActivity activity);
}

less复制代码@Module(subcomponents = LoginComponent.class)
public class SubComponentsModule {
}
--------------------------------------------
@Component(modules = {NetworkModule.class,SubComponentsModule.class})
@Singleton
public interface ApplicationComponent {
    UserRepository userRepository();
    LoginComponent.Factory loginComponent();
}
--------------------------------------------
@Subcomponent
public interface LoginComponent {
    @Subcomponent.Factory
    interface Factory{
        LoginComponent create();
    }
    void inject001(LoginActivity activity);
}

scala复制代码public class LoginActivity extends Activity {
    1、持有子组件引用，保证相同生命周期
    LoginComponent loginComponent;

    2、@Inject 字段注入
    @Inject
    LoginViewModel loginViewModel;

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);

        3、创建子组件实例
        loginComponent = ((MyApplication) getApplicationContext())
                                .appComponent.loginComponent().create();
        4、注入
        loginComponent.inject(this);
        ...
    }
}

java复制代码@Subcomponent
public interface LoginComponent {

    @Subcomponent.Factory
    interface Factory {
        LoginComponent create();
    }

    void inject001(LoginActivity loginActivity);
    void inject002(LoginUsernameFragment fragment);
}

less复制代码@Scope
@Retention(RetentionPolicy.RUNTIME)
public @interface ActivityScope {}

@ActivityScope
@Subcomponent
public interface LoginComponent { ... }

@ActivityScope
public class LoginViewModel {
    private final UserRepository userRepository;

    @Inject
    public LoginViewModel(UserRepository userRepository) {
        this.userRepository = userRepository;
    }
}

scala复制代码public class LoginActivity extends Activity {
    1、持有子组件引用，保证相同生命周期
    LoginComponent loginComponent;

    2、@Inject 字段注入
    @Inject
    FakeLoginViewModel loginViewModel;
}

less复制代码@Singleton
@Component(modules = {FakeNetworkModule.class, SubcomponentsModule.class})
public interface TestApplicationComponent extends ApplicationComponent {
}

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

yaml复制代码spring:
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest

java复制代码@RabbitListener(bindings = @QueueBinding(value = @Queue("DirectA"),key = "ONE",exchange = @Exchange(name = "DirectExchange", type = ExchangeTypes.DIRECT)))
public void processA(String message) {
    logger.info("------> DirectA 发送接收成功  :{}<-------", message);
}

java复制代码rabbitTemplate.convertAndSend("DirectExchange", "ONE", "发送消息 :" + msg);

java复制代码    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("FanoutA"),
            key = "ONE",
            exchange = @Exchange(name = "FanoutExchange", type = ExchangeTypes.FANOUT)
    ))
    public void processA(String message) {
        logger.info("------> FanoutA 发送接收成功  :{}<-------", message);
    }

java复制代码    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("DirectA"),
            key = "ONE",
            exchange = @Exchange(name = "DirectExchange", type = ExchangeTypes.DIRECT)
    ))
    public void processA(String message) {
        logger.info("------> DirectA 发送接收成功  :{}<-------", message);
    }

java复制代码    @RabbitListener(bindings = @QueueBinding(
            value = @Queue("TopicA"),
            key = "ONE.*",
            exchange = @Exchange(name = "TopicExchange", type = ExchangeTypes.TOPIC)
    ))
    public void processA(String message) {
        logger.info("------> TopicA 发送接收成功  :{}<-------", message);
    }

java复制代码// 以下是一个非Spring发送的完整流程 ,在 SpringBoot 中 , 这些流程被代理了
ConnectionFactory connectionFactory = new CachingConnectionFactory();

AmqpAdmin admin = new RabbitAdmin(connectionFactory);
admin.declareQueue(new Queue("myqueue"));

AmqpTemplate template = new RabbitTemplate(connectionFactory);
template.convertAndSend("myqueue", "foo");

String foo = (String) template.receiveAndConvert("myqueue");

java复制代码
// Direct
new Binding(someQueue, someDirectExchange, "foo.bar");

// Topic
new Binding(someQueue, someTopicExchange, "foo.*");

// Fanout
new Binding(someQueue, someFanoutExchange);

//

java复制代码@Bean
PooledChannelConnectionFactory pcf() throws Exception {
    ConnectionFactory rabbitConnectionFactory = new ConnectionFactory();
    rabbitConnectionFactory.setHost("localhost");
    PooledChannelConnectionFactory pcf = new PooledChannelConnectionFactory(rabbitConnectionFactory);
    pcf.setPoolConfigurer((pool, tx) -> {
        if (tx) {
            // configure the transactional pool
        }
        else {
            // configure the non-transactional pool
        }
    });
    return pcf;
}

// 我们来看一下源码 : 

C- ThreadChannelConnectionFactory
    C- ConnectionWrapper
        private final ThreadLocal<Channel> channels = new ThreadLocal<>();
        private final ThreadLocal<Channel> txChannels = new ThreadLocal<>();
        ?- 内部主要通过2个ThreadLocal 来存储 Channel , 之所以是2个其中一个是为事务通道准备的
        
        
C- PooledChannelConnectionFactory
    C- ConnectionWrapper
        private final ObjectPool<Channel> channels;
        private final ObjectPool<Channel> txChannels;
        ?- ObjectPool 是 CommonPool 的组件


//

java复制代码CachingConnectionFactory connectionFactory = new CachingConnectionFactory("somehost");
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");

Connection connection = connectionFactory.createConnection();


// PS : 注意其中的构造器 ,其提供了不限于一种的构造方式
public CachingConnectionFactory(@Nullable String hostname)
public CachingConnectionFactory(int port)
public CachingConnectionFactory(@Nullable String hostNameArg, int port)
public CachingConnectionFactory(URI uri)
public CachingConnectionFactory(com.rabbitmq.client.ConnectionFactory rabbitConnectionFactory)
private CachingConnectionFactory(com.rabbitmq.client.ConnectionFactory rabbitConnectionFactory,
			boolean isPublisherFactory)
// 以及提供了一个静态构造器
private static com.rabbitmq.client.ConnectionFactory newRabbitConnectionFactory()


// 添加自定义客户端连接属性
?- 访问基础连接工厂 , 设置自定义客户机属性
connectionFactory.getRabbitConnectionFactory().getClientProperties().put("thing1", "thing2");


// 再深入一点 : 
1 内部类 CacheMode : 缓存模式 , 包括 CHANNEL , CONNECTION

// CHANNEL 相关属性
- connectionName : 由 ConnectionNameStrategy 生成的连接的名称
- channelCacheSize : 当前配置的允许空闲的最大通道
- localPort : 连接的本地端口
- idleChannelsTx : 当前空闲的事务通道数(缓存)
- idleChannelsNotTx : 当前空闲的非事务性通道的数量(缓存)
- idleChannelsTxHighWater : 已并发空闲的事务通道的最大数量(缓存)
- idleChannelsNotTxHighWater : 非事务性通道的最大数目已并发空闲(缓存)

// CONNECTION 相关属性
- connectionName:<localPort> : 由 ConnectionNameStrategy 生成的连接的名称
- openConnections : 表示到代理的连接的连接对象的数量
- channelCacheSize : 当前配置的允许空闲的最大通道
- connectionCacheSize : 允许空闲的当前配置的最大连接
- idleConnections : 当前空闲的连接数
- idleConnectionsHighWater : 已并发空闲的最大连接数
- idleChannelsTx:<localPort> : 此连接当前空闲的事务通道数(缓存)
- idleChannelsNotTx:<localPort> : 此连接当前空闲的非事务性通道数(缓存)
- idleChannelsTxHighWater:<localPort> : 已并发空闲的事务通道的最大数量(缓存)
- idleChannelsNotTxHighWater:<localPort> : 非事务性通道的最大数目已并发空闲(缓存)

java复制代码void send(Message message) throws AmqpException;
void send(String routingKey, Message message) throws AmqpException;
void send(String exchange, String routingKey, Message message) throws AmqpException;

// 同样 , 在业务中我们可以使用更合理的发送方式
RabbitTemplate template = new RabbitTemplate(); 
template.setRoutingKey("queue.helloWorld"); 

// Template 同样可以定制 , 例如
template.setConfirmCallback(...);

// 这里可以定制配置
MessageProperties properties = new MessageProperties();
//... 省略
template.send(new Message("Hello World".getBytes(), properties));




// 同样 , 可以对 Message 进行详细处理
Message message = MessageBuilder.withBody("foo".getBytes())
    .setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
    .setMessageId("123")
    .setHeader("bar", "baz")
    .build();
public static MessageBuilder withBody(byte[] body) 
    ?- 生成器创建的消息有一个主体，它是对参数的直接引用
    - return new MessageBuilder(body);
public static MessageBuilder withClonedBody(byte[] body) 
    ?- Arrays 新建了一个数组
    - return new MessageBuilder(Arrays.copyOf(body, body.length));
public static MessageBuilder withBody(byte[] body, int from, int to) 
    - return new MessageBuilder(Arrays.copyOfRange(body, from, to));
public static MessageBuilder fromMessage(Message message) 
    - return new MessageBuilder(message);
public static MessageBuilder fromClonedMessage(Message message) 
    - byte[] body = message.getBody();
    - return new MessageBuilder(Arrays.copyOf(body, body.length), message.getMessageProperties());
    
    

// MessageProperties 扩展方式
MessageProperties props = MessagePropertiesBuilder.newInstance()
    .setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN)
    .setMessageId("123")
    .setHeader("bar", "baz")
    .build();
    
public static MessagePropertiesBuilder newInstance() 
    ?-  使用默认值初始化新的消息属性对象
public static MessagePropertiesBuilder fromProperties(MessageProperties properties) 
    ?- return new MessagePropertiesBuilder(properties) : 通过传入的 properties 创建
public static MessagePropertiesBuilder fromClonedProperties(MessageProperties properties) 
    ?- 内部进行了 builder.copyProperties(properties);

java复制代码Message receive() throws AmqpException;
Message receive(String queueName) throws AmqpException;
Message receive(long timeoutMillis) throws AmqpException;
Message receive(String queueName, long timeoutMillis) throws AmqpException;

Object receiveAndConvert() throws AmqpException;
Object receiveAndConvert(String queueName) throws AmqpException;
Object receiveAndConvert(long timeoutMillis) throws AmqpException;
Object receiveAndConvert(String queueName, long timeoutMillis) throws AmqpException;

<R, S> boolean receiveAndReply(ReceiveAndReplyCallback<R, S> callback)throws AmqpException;
<R, S> boolean receiveAndReply(String queueName, ReceiveAndReplyCallback<R, S> callback) throws AmqpException;
<R, S> boolean receiveAndReply(ReceiveAndReplyCallback<R, S> callback,String replyExchange, String replyRoutingKey) throws AmqpException;
<R, S> boolean receiveAndReply(String queueName, ReceiveAndReplyCallback<R, S> callback,String replyExchange, String replyRoutingKey) throws AmqpException;
<R, S> boolean receiveAndReply(ReceiveAndReplyCallback<R, S> callback,ReplyToAddressCallback<S> replyToAddressCallback) throws AmqpException;
<R, S> boolean receiveAndReply(String queueName, ReceiveAndReplyCallback<R, S> callback,ReplyToAddressCallback<S> replyToAddressCallback) throws AmqpException;

java复制代码@EnableRabbit

java复制代码
// Exchange 接口表示 AMQP Exchange，即消息生成器发送到的内容。
// 代理虚拟主机中的每个 Exchange 都有唯一的名称以及一些其他属性
public interface Exchange {
    String getName();
    // direct、 topic、 fanout 和 header
    String getExchangeType();
    boolean isDurable();
    boolean isAutoDelete();
    Map<String, Object> getArguments();
}

// Queue 类表示消息使用者从中接收消息的组件。
// 与各种 Exchange 类一样，我们的实现旨在作为这种核心 AMQP 类型的抽象表示
public class Queue  {
    private final String name;
    private volatile boolean durable;
    private volatile boolean exclusive;
    private volatile boolean autoDelete;
    private volatile Map<String, Object> arguments;
    public Queue(String name) {
        this(name, true, false, false);
    }
}

java复制代码// 默认缓存大小
private static final int DEFAULT_CHANNEL_CACHE_SIZE = 25;
private static final String DEFAULT_DEFERRED_POOL_PREFIX = "spring-rabbit-deferred-pool-";
// channel 超时时间
private static final int CHANNEL_EXEC_SHUTDOWN_TIMEOUT = 30;


// 核心类
private Channel getChannel(ChannelCachingConnectionProxy connection, boolean transactional) {
    Semaphore permits = null; // 发现了一个神奇的工具类 , 可以理解为共享锁
    if (this.channelCheckoutTimeout > 0) {
        permits = obtainPermits(connection);
    }
    LinkedList<ChannelProxy> channelList = determineChannelList(connection, transactional);
    ChannelProxy channel = null;
    if (connection.isOpen()) {
        channel = findOpenChannel(channelList, channel);
    }
    if (channel == null) {
        try {
        
            channel = getCachedChannelProxy(connection, channelList, transactional);
        }catch (RuntimeException e) {
            //Semaphore 的释放 , 这是一个多线程信号量工具
        }
       return channel;
}


private ChannelProxy findOpenChannel(LinkedList<ChannelProxy> channelList,ChannelProxy channelArg) {
    ChannelProxy channel = channelArg;
    synchronized (channelList) {
        while (!channelList.isEmpty()) {
            // 此处训话获取 channelList 中第一个 channel , 并且从集合中移除
            channel = channelList.removeFirst();
                if (channel.isOpen()) {
                    break;
                } else {
                    cleanUpClosedChannel(channel);
                    channel = null;
                }
            }
        }
        return channel;
}

// CachingConnectionFactory

java复制代码// 方法 declareExchanges : 
// 声明了一个 exchange , 其配置主要来源于注解或配置
Map<String, Object> arguments = exchange.getArguments();
channel.exchangeDeclare(
    exchange.getName(), 
    DELAYED_MESSAGE_EXCHANGE, 
    exchange.isDurable(),
    exchange.isAutoDelete(), 
    exchange.isInternal(), 
    arguments
);

//-----------------
// 方法 : declareQueues 声明了一个 Queue
// 方法 : declareBindings 声明了绑定关系

java复制代码
// 节点一 : 先建立 Channel , 再通过 Callback 方法回调建立 Exchange
M- declareExchange(final Exchange exchange)
    // 注意 , 这里 excute 的参数是一个 ChannelCallback , 其目的是为了channle 建立后的回调
    this.rabbitTemplate.execute(channel -> {
        declareExchanges(channel, exchange);
        return null;
    });
    
M- declareExchanges
    ?- 该方法允许一次性传入多个 Exchange 
    - 先准备arguments 属性
    - 调用 channel.exchangeDeclare 实现声明操作

    
C- RabbitTemplate
    M- execute : RabbitTemplate 通过 retryTemplate 进行重试调用
        ?- 注意 doExecute , Channel 的建立主要还是在 RabbitTemplate 中完成的
        ?- 在 doExecute 中再 Callback 之前 RabbitAdmin 里面的操作 , 类似于一个回调
        - return this.retryTemplate.execute(
            (RetryCallback<T, Exception>) context -> doExecute(action, connectionFactory),
            (RecoveryCallback<T>) this.recoveryCallback);
            
// 节点二 : 删除 Exchange
M- deleteExchange(final String exchangeName) : 通过名字删除 Exchange
    - 先判断是否为默认 Exchange , 名称为""
    - 再直接通过channel.exchangeDelete 删除
    

// 节点三 : 建立 Queue , 和Exchange 同理
M- declareQueue(final Queue queue)
    - 同理 , 先建立 Channel 后回调
    - 调用  declareQueues(channel, queue); 完成Queue 声明
        - channel.queueDeclare : 通过 channel 对象完成 Queue 声明
        
M- declareQueue()
    ?- 声明一个服务器命名的独占的、autodelete的、非持久的队列
    ?- 这里是没有传入 Queue 对象的 , 直接创建一个 Chhenl , 然后创建一个 Queue
    -  this.rabbitTemplate.execute(Channel::queueDeclare)
        
M- deleteQueue(final String queueName)
    ?- 同样的 , 先创建一个 Channel , 再删除
    
M-  deleteQueue(final String queueName, final boolean unused, final boolean empty)
    ?- 更详细的删除方式 , 包括是否使用 , 是否为空来判断是否删除
    
M- purgeQueue 
    ?- 清除队列，可以选择不等待清除发生
    
    
// 节点四 : Binding 处理
M- declareBinding
    - 先建立 Channel 
    - declareBindings(channel, binding);
        - channel.queueBind(binding.getDestination(), binding.getExchange(), binding.getRoutingKey(),binding.getArguments());
        - channel.exchangeBind(binding.getDestination(), binding.getExchange(), binding.getRoutingKey(),binding.getArguments());
        
        
        
M- removeBinding(final Binding binding)
    - channel.queueUnbind(binding.getDestination(), binding.getExchange(), binding.getRoutingKey(),binding.getArguments());
    - channel.exchangeUnbind(binding.getDestination(), binding.getExchange(), binding.getRoutingKey(),binding.getArguments());
                                                

M- QueueInformation getQueueInfo(String queueName)
    - DeclareOk declareOk = channel.queueDeclarePassive(queueName);
    - return new QueueInformation(declareOk.getQueue(), declareOk.getMessageCount(),declareOk.getConsumerCount());

java复制代码// 这种方式的体现主要在 Consumer 中, 以BlockingQueueConsumer 为例

// Step 1 : 创建 Consumer 的时候传入 isChannelTransacted
consumer = new BlockingQueueConsumer(getConnectionFactory(), getMessagePropertiesConverter(),
    this.cancellationLock, getAcknowledgeMode(), isChannelTransacted(), actualPrefetchCount,
    isDefaultRequeueRejected(), getConsumerArguments(), isNoLocal(), isExclusive(), queues);

// Step 2 : 最终在相关方法中调用
M- rollbackOnExceptionIfNecessary
    if (this.transactional) -> RabbitUtils.rollbackIfNecessary(this.channel);
    
    
// 事务的声明 : 
1 例如可以使用Spring 事务模型

// 事务的原理 : 
TODO

java复制代码@Bean
public RabbitTransactionManager rabbitTransactionManager() {
    return new RabbitTransactionManager(connectionFactory);
}

@Bean
public SimpleMessageListenerContainer messageListenerContainer() {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(rabbitConnectionFactory());
        container.setTransactionManager(rabbitTransactionManager());
        container.setChannelTransacted(true);
        container.setQueueName("some.queue");
        container.setMessageListener(exampleListener());
        return container;
}

// 深入一下用法 : 
C- SimpleMessageListenerContainer
    M- doReceiveAndExecute
        ?- 其中在一个 Catch 中能看到对 transactionManager 的调用
        ... catch (Exception ex) {
            - 判断是否有transactionManager
            - 判断是否开启了Callback
                RabbitResourceHolder resourceHolder = (RabbitResourceHolder)TransactionSynchronizationManager.getResource(getConnectionFactory());
                if (resourceHolder != null) {
                    consumer.clearDeliveryTags();
                } else {
                    consumer.rollbackOnExceptionIfNecessary(ex);
                }

java复制代码 @RabbitListener(bindings = @QueueBinding(
        value = @Queue(value = "myQueue", durable = "true"),
        exchange = @Exchange(value = "auto.exch", ignoreDeclarationExceptions = "true"),
        key = "orderRoutingKey")
  )

java复制代码// 这里 Queue 没有设置任何东西
  @RabbitListener(bindings = @QueueBinding(
        value = @Queue,
        exchange = @Exchange(value = "auto.exch"),
        key = "invoiceRoutingKey")
  )

java复制代码@RabbitListener(queues = "myQueue")
public void processOrder(Order order, @Header("order_type") String orderType)

java复制代码@RabbitListener(queues = { "queue1", "queue2" } )

// 另外还可以使用 SPEL 表达式
@RabbitListener(queues = "#{'${property.with.comma.delimited.queue.names}'.split(',')}" )

java复制代码@SendTo("#{environment['my.send.to']}")


// 使用Bean 的方式
@RabbitListener(queues = "test.sendTo.spel")
@SendTo("#{spelReplyTo}")
public String capitalizeWithSendToSpel(String foo) {
    return foo.toUpperCase();
}

@Bean
public String spelReplyTo() {
    return "test.sendTo.reply.spel";
}

// SPEL
@SendTo("!{'some.reply.queue.with.' + result.queueName}")

// Spring Enviroment
@SendTo("#{environment['my.send.to']}")

java复制代码@RabbitListener(queues = "q1", messageConverter = "delegating",
        replyContentType = "application/json")

java复制代码// RabbitListenerAnnotationBeanPostProcessor


// Step Start : postProcessAfterInitialization
// postProcessAfterInitialization 来自于 BeanPostProcessor , 他会在自定义初始化之前调用
public Object postProcessAfterInitialization(final Object bean, final String beanName) throws BeansException {
    // 获取 ServletWebServerFactoryConfiguration
    Class<?> targetClass = AopUtils.getTargetClass(bean);
    // 获取每个类的 RabbitListener Metadata
    final TypeMetadata metadata = this.typeCache.computeIfAbsent(targetClass, this::buildMetadata);
    for (ListenerMethod lm : metadata.listenerMethods) {
        for (RabbitListener rabbitListener : lm.annotations) {
            // 如果存在注解 , 则处理
            processAmqpListener(rabbitListener, lm.method, bean, beanName);
        }
    }
    if (metadata.handlerMethods.length > 0) {
        processMultiMethodListeners(metadata.classAnnotations, metadata.handlerMethods, bean, beanName);
    }
    return bean;
}



// Step : findListenerAnnotations 扫描所有的注解
private RabbitListenerAnnotationBeanPostProcessor.TypeMetadata buildMetadata(Class<?> targetClass) {
        // 此处会对每个类都检测是否存在 @RabbitListener
        Collection<RabbitListener> classLevelListeners = findListenerAnnotations(targetClass);
        final boolean hasClassLevelListeners = classLevelListeners.size() > 0;
        final List<RabbitListenerAnnotationBeanPostProcessor.ListenerMethod> methods = new ArrayList<>();
        final List<Method> multiMethods = new ArrayList<>();
        ReflectionUtils.doWithMethods(targetClass, method -> {
            Collection<RabbitListener> listenerAnnotations = findListenerAnnotations(method);
            if (listenerAnnotations.size() > 0) {
                // 如果存在注解 , 则添加到集合中
                methods.add(new RabbitListenerAnnotationBeanPostProcessor.ListenerMethod(method,
                        listenerAnnotations.toArray(new RabbitListener[listenerAnnotations.size()])));
            }
            if (hasClassLevelListeners) {
                RabbitHandler rabbitHandler = AnnotationUtils.findAnnotation(method, RabbitHandler.class);
                if (rabbitHandler != null) {
                    multiMethods.add(method);
                }
            }
        }, ReflectionUtils.USER_DECLARED_METHODS);
        if (methods.isEmpty() && multiMethods.isEmpty()) {
            return RabbitListenerAnnotationBeanPostProcessor.TypeMetadata.EMPTY;
        }
        // 构建一个 TypeMetadata
        return new RabbitListenerAnnotationBeanPostProcessor.TypeMetadata(
                methods.toArray(new RabbitListenerAnnotationBeanPostProcessor.ListenerMethod[methods.size()]),
                multiMethods.toArray(new Method[multiMethods.size()]),
                classLevelListeners.toArray(new RabbitListener[classLevelListeners.size()]));
}

java复制代码// processAmqpListener
protected void processAmqpListener(RabbitListener rabbitListener, Method method, Object bean, String beanName) {
    // 如果是代理 , 返回最终代理的方法
    Method methodToUse = checkProxy(method, bean);
    // 构建了一个 MethodRabbitListenerEndpoint , 该单元用于处理这个端点的传入消息
    MethodRabbitListenerEndpoint endpoint = new MethodRabbitListenerEndpoint();
    endpoint.setMethod(methodToUse);
    // Listener 注解流程
    processListener(endpoint, rabbitListener, bean, methodToUse, beanName);
}


// 我们继续深入 , 很长的一段 ,我们省略掉其中无意义的一部分 :
protected void processListener(MethodRabbitListenerEndpoint endpoint, RabbitListener rabbitListener, Object bean,
    	Object target, String beanName) {

    // endpoint 注入属性 
    endpoint.setBean(bean);
    endpoint.setMessageHandlerMethodFactory(this.messageHandlerMethodFactory);
    endpoint.setId(getEndpointId(rabbitListener));
    // 这里仅仅只是对 QueueName 进行校验和解析
    endpoint.setQueueNames(resolveQueues(rabbitListener));
    // 处理监听器
    endpoint.setConcurrency(resolveExpressionAsStringOrInteger(rabbitListener.concurrency(), "concurrency"));
    endpoint.setBeanFactory(this.beanFactory);
    //异常处理
    endpoint.setReturnExceptions(resolveExpressionAsBoolean(rabbitListener.returnExceptions()));
    
    Object errorHandler = resolveExpression(rabbitListener.errorHandler());
    if (errorHandler instanceof RabbitListenerErrorHandler) {
    	endpoint.setErrorHandler((RabbitListenerErrorHandler) errorHandler);
    }
    else if (errorHandler instanceof String) {
    	String errorHandlerBeanName = (String) errorHandler;
    	if (StringUtils.hasText(errorHandlerBeanName)) {
			endpoint.setErrorHandler(this.beanFactory.getBean(errorHandlerBeanName, RabbitListenerErrorHandler.class));
    	}
    }
    else {
    	throw new IllegalStateException("error handler mut be a bean name or RabbitListenerErrorHandler, not a "
        	+ errorHandler.getClass().toString());
    }
    String group = rabbitListener.group();
    if (StringUtils.hasText(group)) {
    	Object resolvedGroup = resolveExpression(group);
    	if (resolvedGroup instanceof String) {
			endpoint.setGroup((String) resolvedGroup);
    	}
    }
    String autoStartup = rabbitListener.autoStartup();
    if (StringUtils.hasText(autoStartup)) {
    	endpoint.setAutoStartup(resolveExpressionAsBoolean(autoStartup));
    }

    endpoint.setExclusive(rabbitListener.exclusive());
    String priority = resolve(rabbitListener.priority());
    if (StringUtils.hasText(priority)) {
		endpoint.setPriority(Integer.valueOf(priority));
    }

    resolveExecutor(endpoint, rabbitListener, target, beanName);
    resolveAdmin(endpoint, rabbitListener, target);
    // ACK 模式我们回头再看看
    resolveAckMode(endpoint, rabbitListener);
    resolvePostProcessor(endpoint, rabbitListener, target, beanName);
    RabbitListenerContainerFactory<?> factory = resolveContainerFactory(rabbitListener, target, beanName);

    this.registrar.registerEndpoint(endpoint, factory);
}

// 注册 Endpoint
public void registerEndpoint(RabbitListenerEndpoint endpoint,
			@Nullable RabbitListenerContainerFactory<?> factory) {
    AmqpListenerEndpointDescriptor descriptor = new AmqpListenerEndpointDescriptor(endpoint, factory);
    synchronized (this.endpointDescriptors) {
        if (this.startImmediately) { 
            this.endpointRegistry.registerListenerContainer(descriptor.endpoint, // NOSONAR never null
            resolveContainerFactory(descriptor), true);
        }else {
            this.endpointDescriptors.add(descriptor);
        }
    }
}

java复制代码
// Step 1 : RabbitListenerEndpointRegistry Listener 启动
public void start() {
    for (MessageListenerContainer listenerContainer : getListenerContainers()) {
        startIfNecessary(listenerContainer);
    }
}

// 问题一 : getListenerContainers 的来源 >>
RabbitListenerEndpointRegistry  -->  Map<String, MessageListenerContainer>
// 注意 , 这里就和上文的扫描匹配上了 , 由 registerEndpoint 调用
public void registerListenerContainer(RabbitListenerEndpoint endpoint, RabbitListenerContainerFactory<?> factory,
				boolean startImmediately) {

    String id = endpoint.getId();
    synchronized (this.listenerContainers) {
        MessageListenerContainer container = createListenerContainer(endpoint, factory);
        this.listenerContainers.put(id, container);
        if (StringUtils.hasText(endpoint.getGroup()) && this.applicationContext != null) {
            List<MessageListenerContainer> containerGroup;
            if (this.applicationContext.containsBean(endpoint.getGroup())) {
                containerGroup = this.applicationContext.getBean(endpoint.getGroup(), List.class);
            }else {
                containerGroup = new ArrayList<MessageListenerContainer>();
                this.applicationContext.getBeanFactory().registerSingleton(endpoint.getGroup(), containerGroup);
            }
            containerGroup.add(container);
        }
        if (this.contextRefreshed) {
            container.lazyLoad();
        }
        if (startImmediately) {
            startIfNecessary(container);
        }
    }
}

// 扫描时 , 会将 MessageListenerContainer 放入 Map , 用于后续使用
// Step 2 : RabbitListenerEndpointRegistry Listener 运行
private void startIfNecessary(MessageListenerContainer listenerContainer) {
    if (this.contextRefreshed || listenerContainer.isAutoStartup()) {
        // 主要是 SimpleMessageListenerContainer
        // 继承了 AbstractMessageListenerContainer , start 在该类中
        listenerContainer.start();
    }
}

// Step 3 : start 方法

configureAdminIfNeeded();
// 校验 Queue 情况 ,主要是调用 RabbitAdmin , 校验失败的会 Channel shutdown
//  Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'type' for exchange 'HeaderExchange' in vhost '/': received 'direct' but current is 'headers', class-id=40, method-id=10)
checkMismatchedQueues();
    - 1 判断Exchange 情况并且输出日志 : !isDurable / isAutoDelete / isInfoEnabled
    - 2 判断 Queue 情况并且输出日志 : !isDurable / isAutoDelete / isExclusive / isInfoEnabled(开启日志)
    // 注意 , 这里就是之前苦苦寻找声明的地方 , 也可以理解为这里就开始连接了
    - 3 declareExchanges(channel, exchanges.toArray(new Exchange[exchanges.size()]));
        - channel.exchangeDeclare(exchange.getName(), DELAYED_MESSAGE_EXCHANGE, exchange.isDurable(),exchange.isAutoDelete(), exchange.isInternal(), arguments);
    - 4 declareQueues(channel, queues.toArray(new Queue[queues.size()]));
        - channel.queueDeclare(queue.getName(), queue.isDurable(),queue.isExclusive(), queue.isAutoDelete(), queue.getArguments());
    - 5 declareBindings(channel, bindings.toArray(new Binding[bindings.size()]));
        - channel.queueBind(binding.getDestination(), binding.getExchange(), binding.getRoutingKey(),binding.getArguments());
// 最后调用了一下子类的 doStart , 用于后续扩展
doStart();

java复制代码rabbitTemplate.convertAndSend("DirectExchange", "ONE", "发送消息 :" + msg);

java复制代码
// 中间有一个 Boolean 型 ,我们用伪代码表示
Boolean item1 = RabbitTemplate.this.returnCallback != null|| (correlationData != null && StringUtils.hasText(correlationData.getId()))
Boolean item2 = RabbitTemplate.this.mandatoryExpression.getValue(RabbitTemplate.this.evaluationContext, message, Boolean.class)
Boolean mandatory =  item1 && item2 ;
            

@Override
public void send(final String exchange, final String routingKey,
                    final Message message, @Nullable final CorrelationData correlationData)
                    throws AmqpException {
    execute(channel -> {
        // 核心方法 , 发送消息
        doSend(channel, exchange, routingKey, message,mandatory,correlationData);
        return null;
    }, obtainTargetConnectionFactory(this.sendConnectionFactorySelectorExpression, message));
}

// doSend 中我们同样去掉无关的
public void doSend(Channel channel, String exchangeArg, String routingKeyArg, Message message, // NOSONAR complexity
			boolean mandatory, @Nullable CorrelationData correlationData) throws IOException {

    // 这里准备了 Exchange 参数 和 routingKey 参数
    String exch = exchangeArg;
    String rKey = routingKeyArg;

    // 中间一大段都是对 MessageProperties 进行处理 , MessageProperties 存在 messageToUse 中
   
    sendToRabbit(channel, exch, rKey, mandatory, messageToUse);
    if (isChannelLocallyTransacted(channel)) {
        RabbitUtils.commitIfNecessary(channel);
    }
}

protected void sendToRabbit(Channel channel, String exchange, String routingKey, boolean mandatory,Message message) throws IOException {

    BasicProperties convertedMessageProperties = this.messagePropertiesConverter.fromMessageProperties(message.getMessageProperties(), this.encoding);
    // 这里就是常规的 RabbitMQ Client 操作了 , 到这里 发送就结束了
    // com.rabbitmq.client.impl.ChannelN
    channel.basicPublish(exchange, routingKey, mandatory, convertedMessageProperties, message.getBody());
    
    // basicPublish 核心如下 , channle 的逻辑我们可以未来再深入
    AMQCommand command = new AMQCommand((new Builder()).exchange(exchange).routingKey(routingKey).mandatory(mandatory).immediate(immediate).build(), props, body);
    this.transmit(command);

    
}

java复制代码
// Step 1: SimpleMessageListenerContainer#run
// 其中有这样一段代码
while (isActive(this.consumer) || this.consumer.hasDelivery() || !this.consumer.cancelled()) {
    mainLoop();
}

// Step 2 : 持续监听 ,往里面追溯 , 可以发现有一段这样的代码
M- doReceiveAndExecute
try {
    executeListener(channel, message);
}

-> doExecuteListener
-> invokeListener(channel, message);
-> this.proxy.invokeListener(channel, data) // 到这里实际上还是看不到实际的处理类

// 一路debug , 最终会到这个方法中
protected void actualInvokeListener(Channel channel, Object data) {
    Object listener = getMessageListener();
    if (listener instanceof ChannelAwareMessageListener) {
        doInvokeListener((ChannelAwareMessageListener) listener, channel, data);
    }else if (listener instanceof MessageListener) {
			boolean bindChannel = isExposeListenerChannel() && isChannelLocallyTransacted();
			if (bindChannel) {
				RabbitResourceHolder resourceHolder = new RabbitResourceHolder(channel, false);
				resourceHolder.setSynchronizedWithTransaction(true);
				TransactionSynchronizationManager.bindResource(this.getConnectionFactory(),
						resourceHolder);
			}
			try {
				doInvokeListener((MessageListener) listener, data);
			}
			finally {
				if (bindChannel) {
					// unbind if we bound
					TransactionSynchronizationManager.unbindResource(this.getConnectionFactory());
				}
			}
		}
		else if (listener != null) {
			throw new FatalListenerExecutionException("Only MessageListener and SessionAwareMessageListener supported: "
					+ listener);
		}
		else {
			throw new FatalListenerExecutionException("No message listener specified - see property 'messageListener'");
		}
	}

java复制代码// 配置方式
@Bean
public SimpleMessageListenerContainer messageListenerContainer(ConnectionFactory connectionFactory){
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setQueueNames("queueName1","queueName2");
        container.setMessageListener((MessageListener) message -> {
           //....
        });
        return container;
}