MassTransit，直译公共交通， 是由Chris Patterson开发的基于消息驱动的.NET 分布式应用框架，其核心思想是借助消息来实现服务之间的松耦合异步通信，进而确保应用更高的可用性、可靠性和可扩展性。通过对消息模型的高度抽象，以及对主流的消息代理（包括RabbitMQ、ActiveMQ、Kafaka、Azure Service Bus、Amazon SQS等）的集成，大大简化了基于消息驱动的开发门槛，同时内置了连接管理、消息序列化和消费者生命周期管理，以及诸如重试、限流、断路器等异常处理机制，让开发者更好的专注于业务实现。 简而言之，MassTransit实现了消息代理透明化。无需面向消息代理编程进行诸如连接管理、队列的申明和绑定等操作，即可轻松实现应用间消息的传递和消费。
MassTransit本身定位轻量级的服务总线，并支持多种传输方式如：RabbitMQ、Azure Service Bus、ActiveMQ、Amazon SQS、Kafka、Azure Event Hub。消息异常处理：重试配置、重新交付、erro管道、死信管道。分布式事务处理：sagas、Courier。容器支持：.NETcore自身的、autofac、castle windsor等、调度支持：Quartz 、hangfire。更多功能参考官网文档。
编排一系列事件的能力是一个强大的功能，而MassTransit使这成为可能。 saga是由协调器管理的长期事务。saga是由事件发起的，saga编排事件，saga维护整个事务的状态。saga旨在管理分布式事务的复杂性，而不需要锁定和一致性。它们管理状态并跟踪发生部分故障时所需的任何补偿。
RabbitMQ是使用Erlang编写的一个开源的消息队列，本身支持很多的协议：AMQP，XMPP, SMTP, STOMP，也正因如此，它非常重量级，更适合于企业级的开发。同时实现了Broker构架，这意味着消息在发送给客户端时先在中心队列排队。对路由，负载均衡或者数据持久化都有很好的支持。
RabbitMQ无疑是目前最流行的消息队列之一，对各种语言环境的支持也很丰富，作为一个.NET developer有必要学习和了解这一工具。消息队列的使用场景大概有3种：
方法改为 CreateUsingRabbitMq，并且添加 rabbitmq host
消费者的类型包括：普通消费者，saga，saga 状态机，路由活动（分布式追踪），处理器 handlers，工作消费者 job comsumers
DAPP的底层区块链开发平台，就像手机的iOS和Android系统一样，是各种DAPP的潜在生态环境。DApp是源自底层区块链平台生态的各种分布式应用程序，也是区块链世界中的基本服务提供商。Dapp在区块链中，就像应用程序在iOS和Android中一样。
调停者模式是一种行为设计模式，它通过引入一个调停者对象来集中处理一组对象之间的交互。调停者模式的目标是减少对象之间的直接通信，从而降低耦合度，并且使代码更易于维护和扩展。
像activeMQ等消息队列中，我们经常会使用发布订阅模式，但是你有没有想过，客户端时如何及时得到订阅的主题的信息？其实就里就用到了观察者模式。在软件系统中，当一个对象的行为依赖于另一个对象的状态时，观察者模式就相当有用。如果不使用观察者模式提供的通用结构，而需要我们实现类似的功能，想想我们该如何实现，我们只能在另外一个线程不断监听对象所依赖的状态。当然下面的例子都是基于一个进程内观察者模式的举例，你可能会和我当初一样不解，消息队列中的消费者是通过socket进行通信得到订阅的主题的信息。其实还是一回事的，被观察者（主题）里面会维护一个与它有订阅的所有消费者的连接，当被观察者（主题）里面添加一个消息时，就会调用自身的方法，把该消息通过维持的socket发送给所有订阅的消费者。
我是架构精进之路，点击上方“关注”，坚持每天为你分享技术干货，私信我回复“01”，送你一份程序员成长进阶大礼包。
核心是一个任务队列，生产者线程生产任务，并将任务添加到任务队列中，而消费者线程从任务队列中获取任务并执行。该模式有如下优点：
Saga 最初出现在1987年Hector Garcaa-Molrna & Kenneth Salem发表的一篇名为《Sagas》的论文里。其核心思想是将长事务拆分为多个短事务，借助Saga事务协调器的协调，来保证要么所有操作都成功完成，要么运行相应的补偿事务以撤消先前完成的工作，从而维护多个服务之间的数据一致性。举例而言，假设有个在线购物网站，其后端服务划分为订单服务、支付服务和库存服务。那么一次下订单的Saga流程如下图所示：
在这篇文章中，我们将探索如何使用.NET 5中的新source generator特性，使用MediatR库和CQRS模式自动为系统生成API。
系统结构可能会日益变得复杂，对象之间存在大量的相互关联和调用，系统的整体结构容易变为网状结构。在这种情况下，如果需要修改某一个对象，则可能会要跟踪和该对象关联的其他所有对象，并进行处理。耦合越多，修改的地方就会越多。
为了更有效地同步对任何资源的访问，我们可以将条件与任务相关联，让任何线程等待，直到满足某个条件，或者通知其他线程该条件正在满足，以便它们可以解除对自身的阻止。
Redis Stream 是 Redis 5.0 版本引入的一种新的数据类型，它是一个持久化的、可查询的、可扩展的消息队列服务。
ArrayBlockingQueue是一个基于数组的有界阻塞队列。它在创建时需要指定队列的大小，并且这个大小在之后是不能改变的。队列中的元素按照FIFO（先进先出）的原则进行排序。ArrayBlockingQueue是线程安全的，可以在多线程环境下安全地使用。
这些队列都实现了Queue接口或其子接口，可以根据不同的场景和需求选择合适的队列。在并发场景下，应当注意队列的线程安全性以及对并发操作的支持程度。
今天我要给大家分享一些自己日常学习到的一些知识点，并以文字的形式跟大家一起交流，互相学习，一个人虽可以走的更快，但一群人可以走的更远。
关于数据一致性的文章，园子里已经有很多了，如果你还不了解，那么可以通过以下的几篇文章去快速地了解了解，有个感性认识即可。
3.2.0 版本包含许多新功能和改进。本文将重点介绍一些最突出的新功能。有关更改的完整列表，请务必查看发行说明。您还可以观看发布视频，了解 Apache Kafka 3.2.0 中的新功能摘要。
除了提供对SortedSet进行同步包装的方法之外，java.util.Collections还提供了一系列对其他的基础容器进行同步包装的方法，如synchronizedList()方法将基础List包装成线程安全的列表容器，synchronizedMap()方法将基础Map容器包装成线程安全的容器，synchronizedCollection()方法将基础Collection容器包装成线程安全的Collection容器与同步包装方法相对应，java.util.Collections还提供了一系列同步包装类，这些包装类都是其内部类。这些同步包装类的实现逻辑很简单：实现了容器的操作接口，在操作接口上使用synchronized进行线程同步，然后在synchronized的临界区将实际的操作委托给被包装的基础容器。高并发容器： JUC高并发容器是基于非阻塞算法（或者无锁编程算法）实现的容器类，无锁编程算法主要通过CAS（Compare And Swap）+Volatile组合实现，通过CAS保障操作的原子性，通过volatile保障变量内存的可见性。无锁编程算法的主要优点如下： （1）开销较小：不需要在内核态和用户态之间切换进程。 （2）读写不互斥：只有写操作需要使用基于CAS机制的乐观锁， 读读操作之间可以不用互斥。 JUC包中提供了List、Set、Queue、Map各种类型的高并发容器，如ConcurrentHashMap、ConcurrentSkipListMap、ConcurrentSkipListSet、CopyOnWriteArrayList和CopyOnWriteArraySet。在性能上，ConcurrentHashMap通常优于同步的HashMap，ConcurrentSkipListMap通常优于同步的TreeMap。当读取和遍历操作远远大于列表的更新操作时，CopyOnWriteArrayList优于同步的ArrayList。 List：JUC包中的高并发List主要有CopyOnWriteArrayList，对应的基础容器为ArrayList。CopyOnWriteArrayList相当于线程安全的ArrayList，它实现了List接口。在读多写少的场景中，其性能远远高于ArrayList的同步包装容器。 Set：·CopyOnWriteArraySet继承自AbstractSet类，对应的基础容器为HashSet。其内部组合了一个CopyOnWriteArrayList对象，它的核心操作是基于CopyOnWriteArrayList实现的。 ·ConcurrentSkipListSet是线程安全的有序集合，对应的基础容器为TreeSet。它继承自AbstractSet，并实现了NavigableSet接口。ConcurrentSkipListSet是通过ConcurrentSkipListMap实现的。 Map：·ConcurrentHashMap对应的基础容器为HashMap。JDK 6中的ConcurrentHashMap采用一种更加细粒度的“分段锁”加锁机制，JDK 8中采用CAS无锁算法。 ·ConcurrentSkipListMap对应的基础容器为TreeMap。其内部的SkipList（跳表）结构是一种可以代替平衡树的数据结构，默认是按照Key值升序的。 Queue：JUC包中的Queue的实现类包括三类：单向队列、双向队列和阻塞队列。 ·ConcurrentLinkedQueue是基于列表实现的单向队列，按照FIFO（先进先出）原则对元素进行排序。新元素从队列尾部插入，而获取队列元素则需要从队列头部获取。 ·ConcurrentLinkedDeque是基于链表的双向队列，但是该队列不允许null元素。ConcurrentLinkedDeque可以当作“栈”来使用，并且高效地支持并发环境。 ·ArrayBlockingQueue：基于数组实现的可阻塞的FIFO队列。 ·LinkedBlockingQueue：基于链表实现的可阻塞的FIFO队列。 ·PriorityBlockingQueue：按优先级排序的队列。 ·DelayQueue：按照元素的Delay时间进行排序的队列。 ·SynchronousQueue：无缓冲等待队列。
Kafka 由一个或多个节点组成的工作集群，这些节点可以位于不同的数据中心，我们可以在 Kafka 集群的不同节点之间分布数据/负载，并且它天生具有可扩展性、可用性和容错性。
对业务开发来说，无法接触到BufferQueue，甚至不知道BufferQueue是什么东西。对系统来说，BufferQueue是很重要的传递数据的组件，Android显示系统依赖于BufferQueue，只要显示内容到“屏幕”(此处指抽象的屏幕，有时候还可以包含编码器)，就一定需要用到BufferQueue，可以说在显示/播放器相关的领域中，BufferQueue无处不在。即使直接调用Opengl ES来绘制，底层依然需要BufferQueue才能显示到屏幕上。
ArrayBlockingQueue类是一个高效、线程安全的队列实现，它基于数组，提供了快速的元素访问，并支持多线程间的同步操作，作为有界队列，它能有效防止内存溢出，并通过阻塞机制平衡生产者和消费者的速度差异，它还提供了公平性和非公平性策略，满足不同场景下的需求。
我在前段时间写了一篇关于AQS源码解析的文章AbstractQueuedSynchronizer超详细原理解析AbstractQueuedSynchronizer超详细原理解析
并发集合 1 为什么使用并发集合？ 原因主要有以下几点： System.Collections和System.Collections.Generic名称空间中所提供的经典列表、集合和数组都不是线程安全的，若无同步机制，他们不适合于接受并发的指令来添加和删除元素。 在并发代码中使用上述经典集合需要复杂的同步管理，使用起来很不方便。 使用复杂的同步机制会大大降低性能。 NET Framework 4所提供的新的集合尽可能地减少需要使用锁的次数。这些新的集合通过使用比较并交换（compare-and-swap,C
所谓阻塞，简单来说就是当某些条件不满足时，让线程处于等待状态。例如经典的“生产者-消费者”模型，当存放产品的容器满的时候，生产者处于等待状态；而当容器为空的时候，消费者处于等待状态。阻塞队列的概念与该场景类似。
消息的可靠性是 RabbitMQ 的一大特色，那么 RabbitMQ 是如何保证消息可靠性的呢——消息持久化。
中介者模式（Mediator Pattern）是一种行为设计模式，用于通过一个中介对象来封装一系列对象之间的交互，从而使对象之间的耦合度降低，增强系统的可维护性和可扩展性。在Java中，中介者模式是一种常见且有用的模式，可以帮助我们简化对象之间的复杂交互，从而提高代码的可读性和可维护性。本教程将深入介绍Java中的中介者模式，包括其定义、结构、工作原理和实际应用。
在现代应用系统中，缓存是提高性能和减少数据库负载的重要手段之一。然而，缓存的数据在某些情况下可能会过期或者变得无效，因此需要及时进行清理。在复杂的应用系统中，可能有多个系统、多个模块产生缓存清理需求，而这些系统、模块之间的清理任务需要高效的协作，以避免数据竞争和资源浪费的问题。
在上一篇介绍消费者时有简单的提到过WorkerPool，本篇将对WorkerPool进行详细地介绍。 1. 介绍下WorkerPool: workProcessors:事件处理器，每个workProc
RabbitMQ允许你为messages和queues设置TTL（存活时间）。这可以使用可选的queue 参数或策略来完成（建议使用后一个选项）。
线程在同时争抢cpu资源的时候，如果没有设置优先级执行顺序是比较乱的。如果设置了优先级则当线程碰撞在一起的时候，优先级高的就会先执行。
Java7中加入了JSR 166y规范对集合类和并发类库的改进。其中的一项是增加了接口TransferQueue和其实现类LinkedTransferQueue。
一般调用disruptor的hadleEventsWith方法添加event handler处理消费到的event。
RocketMQ 支持两种消息模式：集群消费（ Clustering ）和广播消费（ Broadcasting ）。
生产者的任务就是将消息添加到Redis的Sorted Set中。首先，需要计算出消息添加到Redis的SlotKey，如果发送方指定了消息的slotBasis，则计算slotBasis的CRC32值，CRC32值对槽数量进行取模得到槽序号，SlotKey设计为#{topic}_#{index}，其中#{}表示占位符。然后，不同类型的消息有不同的添加方式，因此分布式讲述的三种消息类型的添加过程
数据流 在当今的数据环境中，没有一个系统可以提供所有必需的观点来提供真正的洞察力。从数据中获取完整含义需要混合来自多个来源的大量信息。 与此同时，我们不耐烦地立即获得答案;如果洞察时间超过10毫秒，那么该值就会丢失 - 高频交易，欺诈检测和推荐引擎等应用程序不能等待。这通常意味着在数据进入记录数据库之前分析数据的流入。为数据丢失增加零容忍，挑战变得更加艰巨。 Kafka和数据流专注于从多个消防软管摄取大量数据，然后将其路由到需要它的系统 - 过滤，汇总和分析途中。 本文介绍了Apache Kafka，
继之前《Kafka运维篇之初识Streams Messaging Manager》、《Kafka运维篇之使用SMM监控Kafka集群》和《Kafka运维篇之使用SMM预警策略管理Kafka预警》之后。我们今天介绍使用SMM来监控Kafka端到端的延迟。
1 基本概念 SOA 公共的业务被拆分出来，形成可共用的服务，最大程度地保障代码和逻辑的复用，避免重复建设，这种设计称为SOA。 路由 SOA架构中，服务消费者通过服务名称，在众多服务中心找到要调用的服务的地址列表，称为服务的路由。 负载均衡 对于负载高的服务，一般有多台服务器组成的集群，当请求到来时，为了将请求均衡的分配到后端服务器，负载均衡程序将从服务对应的地址列表中，通过相应的负载均衡算法和法则，选取一台服务器进行访问，这个过程称为服务的负载均衡 服务配置中心 当服务越来越多，规模变大，单靠人
Kafka 的基本数据单元被称为 message(消息)，为减少网络开销，提高效率，多个消息会被放入同一批次 (Batch) 中后再写入。
在软件开发过程中，会遇见很多的问题场景，对于经常遇到的问题场景，一些大佬总结出一些针对特有场景的固有套路，按照这些套路，将帮助我们将问题简单化，条理清楚的解决问题，这也是设计模式的初衷；
TransferQueue(java7引入)继承了BlockingQueue（BlockingQueue又继承了Queue）并扩展了一些新方法。生产者会一直阻塞直到所添加到队列的元素被某一个消费者所消费（不仅仅是添加到队列里就完事）。
ConsumerInterceptor是Kafka中的一个重要组件，它允许开发人员在Kafka消费者端拦截和修改消息的处理过程。ConsumerInterceptor可以用于实现各种功能，从消息监控到数据转换和错误处理，为开发人员提供了更大的灵活性和可定制性。
领取专属 10元无门槛券
手把手带您无忧上云
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM
