Akka streams:将新的发布者/订阅者附加到Flow
Akka Streams是一种用于构建可扩展、高吞吐量和弹性的流处理应用程序的工具包。它是Akka框架的一部分，提供了一种声明式的编程模型，用于处理和转换数据流。
Akka Streams的核心概念是流（Stream）、流处理器（Flow）和源（Source）/汇（Sink）。流表示一系列的数据元素，可以是有限的或无限的。流处理器是一种将输入流转换为输出流的组件，可以进行各种操作，如过滤、映射、合并等。源是一个产生数据流的组件，而汇是一个接收数据流的组件。
Akka Streams的优势包括：
- 可扩展性：Akka Streams使用异步非阻塞的处理模型，可以轻松地处理大量的并发流。
- 高吞吐量：通过利用并行处理和异步IO，Akka Streams可以实现高吞吐量的数据处理。
- 弹性：Akka Streams提供了一套弹性机制，可以处理故障和异常情况，并自动进行恢复和重试。
- 声明式编程模型：Akka Streams使用简洁的声明式编程模型，使得开发者可以更容易地理解和维护流处理逻辑。
Akka Streams的应用场景包括：
- 实时数据处理：Akka Streams可以用于处理实时数据流，如日志处理、实时监控等。
- 数据转换和过滤：Akka Streams提供了丰富的操作符，可以进行数据转换、过滤和聚合等操作。
- 批处理：Akka Streams可以用于批处理任务，如数据清洗、ETL等。
- 分布式计算：Akka Streams可以与Akka集群和Akka分布式数据一起使用，实现分布式计算任务。
腾讯云提供了一系列与流处理相关的产品和服务，例如：
- 腾讯云消息队列 CMQ：提供了可靠的消息传递服务，可以与Akka Streams结合使用，实现异步消息处理。
- 腾讯云流计算 SCF：提供了无服务器的计算服务，可以用于处理实时数据流。
- 腾讯云数据流水线 DataWorks：提供了数据集成和数据处理的服务，可以与Akka Streams一起使用，实现数据流的ETL和分析。
更多关于Akka Streams的信息和使用方法，可以参考腾讯云的官方文档：Akka Streams - 腾讯云
相关·内容
1回答
我试图写一个卡夫卡消费者到websocket流使用反应性- Kafka，akka-http和akka-流。
val publisherActor = actorSystem.actorOf(CommandPublisher.props)
val publisher = ActorPublisher[String](publisherActor)
val commandSource = Source.fromPublisher(publisher) map toMessage
def toMessage(c: String): Message = TextMessage.Stri
浏览 1提问于2016-04-01得票数 6
1回答
在探索新API akka.actor.typed，中的分布式发布订阅时，我们看到分布式发布订阅是通过使用一个参与者akka.actor.typed.pubsub.Topic.表示每个发布子主题来实现的。为了模拟同样的情况，创建了一个本地akka集群，其中3个种子节点，其中一个充当发布者，另外两个是使用者。在使用时，我们看到两个订户都在接收相同的数据，即去重复不会发生。请查找发行者的示例代码：
object PubApp {
def pubsubImplementor() = {
val pubsubGurdian = Behaviors.setup[Unit] { (conte
浏览 9提问于2022-03-22得票数 1
1回答
我是新使用akka streams kafka (和akka streams一般)。我正在尝试构建一个图表，以便将消息发布到不同的主题。如何将生产者作为流进行连接，以便提交已处理的消息？我试过使用Producer.flow，但我无法获取commitScaladsl
object TestFoo {
import akka.kafka.ProducerMessage.Message
implicit val system = ActorSystem("test-kafka")
implicit val materializer = ActorMaterializer(
浏览 12提问于2017-08-14得票数 0
回答已采纳
我希望用Flink替换我的基于Akka Streams的流处理器的一部分。目前是否可以使用Akka Streams作为Flink的源，然后在同一代码库中使用Flink作为Akka Streams的源？
Akka Streams的当前流量如下：
// Kafka Source -> Read Flow involving Azure cosmosDB -> Evaluate Flow -> Write Flow -> Logger Flow -> Sink
lazy private val converterGraph: Graph[ClosedShape.t
浏览 2提问于2019-04-24得票数 1
2回答
动态合并Akka流
、、、、
我正在尝试使用Akka streams以以下方式构建一个pub sub总线：
发布者为主题添加源流，订阅者指定主题并获取该主题的所有内容。但是，主题可以由多个发布者发布，发布者和订阅者都可以在任何时候加入。
我的想法是合并所有的源，然后将过滤后的源返回给订阅者。
然而，由于发布者可以在任何点加入，所以可以在进行订阅之后添加源，订阅者需要像主题的任何其他发布数据一样从源中获取数据。
有没有一种方法可以动态地管理流到源的合并，从而满足以下条件：
publish(topic: String, messages: Source[T])
subscribe(topic: String): Source[
浏览 1提问于2019-07-05得票数 2
我有以下代码，它不让连接打开到websocket服务器：
import akka.Done
import akka.actor.{Actor, ActorLogging, Props}
import akka.http.scaladsl.model.StatusCodes
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.ws.{Message, TextMessage, WebSocketRequest}
import akka.stream.{ActorMaterializer, Materializer}
imp
浏览 0提问于2019-04-16得票数 0
回答已采纳
我在尝试Java 9中的一些新特性，所以我做了一个测试，让一个发布者以给定的速率发出数字。我还实现了一个订阅服务器来监听这些出版物，并将它们打印到控制台。
尽管我可能不完全理解如何使用这个getLastItem()，因为Api方法不打印任何内容，而Api只返回0。
唯一起作用的部分似乎是正确初始化lastItem变量的onSubscribe()。
@Test
public void testReactiveStreams(){
//Create Publisher
SubmissionPublisher<Integer> publisher = new Submis
浏览 4提问于2017-03-15得票数 5
1回答
我需要创建一个具有以下接口的函数：
import akka.kafka.scaladsl.Consumer.Control
object ItemConversionFlow {
def build(config: StreamConfig): Flow[Item, OtherItem, Control] = {
// Implementation goes here
}
我的问题是，我不知道如何定义流，使其适合上面的接口。
当我做这样的事情时
val flow = Flow[Item]
.map(item => doConversion(item)
.fil
浏览 5提问于2018-01-17得票数 2
我需要为发布到DistributedPubSubMediator主题的消息提供至少一次的传递保证。
我查看了DistributedPubSubMediator.scala，在TopicLike特征(Akka 2.4.6)中可以看到以下内容：
trait TopicLike extends Actor {
var subscribers = Set.empty[ActorRef]
def defaultReceive: Receive = {
case msg ?
subscribers foreach { _ forward msg }
}
}
然而，我找不到
浏览 0提问于2016-05-27得票数 3
我正在寻找akka.stream.scaladsl.Source构造方法，它允许我简单地从不同的代码位置发出下一个值(例如监视系统事件)。
我需要一些类似于Promise的东西。承诺向Future发出单个值。我需要向Source发出多个值。
就像monix.reactive.subjects.BehaviorSubject.onNext(_)
我不太关心背压。
目前，我已经使用monix & akka-streams (下面的代码)实现了这一点，但我希望应该只有akka-streams解决方案：
import akka.stream.scaladsl.{Flow,
浏览 4提问于2020-03-05得票数 1
回答已采纳
2回答
我通过了一个关于中的新Flow相关接口的Java9。下面的示例代码：
public class MySubscriber<T> implements Subscriber<T> {
private Subscription subscription;
@Override
public void onSubscribe(Subscription subscription) {
this.subscription = subscription;
subscription.request(1); //a value of
浏览 0提问于2017-07-19得票数 11
回答已采纳
我是akka streams的新手，akka输入了我的代码，与经典的akka一起工作很好，这就是我要做的。
class FlowActor extends Actor{
val log = LoggerFactory.getLogger(this.getClass)
def receive: Receive = {
case word: String =>
val replyTo = word.capitalize
log.info("Actor {}, Capitalizing {}",context.self.path, re
浏览 3提问于2020-06-16得票数 1
回答已采纳
我正在通过这个学习Angular2。
下面的代码突然出现了。为什么"missionAnnounced$“没有变量声明？让任务宣布$= ...下面代码的逻辑是什么？
import { Injectable } from '@angular/core';
import { Subject } from 'rxjs/Subject';
@Injectable()
export class MissionService {
// Observable string sources
private missionAnnouncedSource = n
浏览 0提问于2016-10-30得票数 9
我正在使用一家提供实时流的数据供应商的Java API。我想使用Akka streams来处理这个流。 Java API有一个pub sub设计，其工作原理大致如下： Subscription sub = createSubscription();
sub.addListener(new Listener() {
public void eventsReceived(List events) {
for (Event e : events)
buffer.enqueue(e)
}
浏览 32提问于2021-09-28得票数 1
回答已采纳
2回答
我正在使用Akka HTTP cachedHostConnectionPoolHttps池作为Akka Streams流的一部分发送请求：
private val requestFlow: Flow[(HttpRequest, HelperClass), Either[Error, String], _] =
Http().cachedHostConnectionPoolHttps(BaseUrl).mapAsync(1) {
case (Success(HttpResponse(_, _, entity, _)), _) =>
Unmarshal(entity).to[
浏览 2提问于2020-05-28得票数 2
我有一个产生事件的publisher类，以及一堆想要订阅事件以便运行自己的回调的侦听器类。
所有这些侦听器类都是预先知道的。
让我的所有监听器在应用程序初始化时订阅发布者的最佳方式是什么？我希望所有特定于侦听器的代码都是侦听器类的一部分。我认为侦听器可以在静态块(对象)中订阅发布者，但据我所知，Java延迟加载类，所以不能保证所有的侦听器都会立即被订阅。为了避免这种情况，我可能会做一些事情来强制JVM立即加载我的所有侦听器？
我是不是遗漏了什么更简单的东西？
浏览 3提问于2011-05-04得票数 4
1回答
我正在为一个中型系统测试ZeroMQ作为发布-订阅(服务总线风格)基础架构。我们大约有50个节点，它们都应该是发布者和订阅者。网络是一种星型拓扑，但边缘相互“对话”。我们需要动态发现(不需要对参与者的网络地址进行硬编码)，但也不需要SPOF (单点故障)。
我读过，据我所知，建议的动态发现0MQ方法涉及一个代理节点(XPUB/XSUB)，它转发订阅和发布。我考虑过在我们的系统中使用这样的代理作为中心中介器，然而，我对这种体系结构有以下担忧：(A)代理节点是SPOF -当它发生故障时，整个系统不起作用(B)所有流量，包括数据，都通过代理节点，这意味着延迟和性能问题。
假设我正确理解了发布-订阅白
浏览 9提问于2013-08-20得票数 12
问题：
Need help successfully running test from Reactive Streams TCK (Technology Compatibility Kit)?
SimplePublisher.java (示例)
package aaa.bbb.ccc.jar;
import java.util.Iterator;
import java.util.stream.IntStream;
import org.reactivestreams.FlowAdapters;
import java.util.concurrent.Flow;
public clas
浏览 0提问于2019-08-13得票数 2
回答已采纳
我目前正在构建一个从mongoDb到elasticsearch的数据流解决方案。我的目标是跟踪所有成功传输到elasticsearch的项目。我正在使用akka-streams和elastic4s。目前，进入es的流如下所示
val esSubscriber: BulkIndexingSubscriber[CustomT] = esClient.subscriber[CustomT](
batchSize = batchSize,
completionFn = { () => elasticFinishPromise.success(()); ()},
error
浏览 21提问于2016-07-25得票数 0
我有一个流代码
val source = Source.actorRef[Msg[T]](1000, OverflowStrategy.dropHead)
val flow=source.map(println(_)).to(Sink.onComplete{case _=>println("complete")}).run()
在运行代码之后，我从日志文件中发现，akka创建了两个参与者：
akka://xxxx/flow-0-0-unknown-operation#-969158424]
akka://xxxx/flow-0-1-actorRefSour
浏览 3提问于2016-05-01得票数 2
回答已采纳
我正在试验reactive Streams的Test Compatability Kit (tck)，并且我自己测试了几个Publisher，我想测试一个akka Source。但为了做到这一点，我需要将源(或源+处理器)转换为java.util.concurrent.Flow.Publisher。
@Override
public Flow.Publisher<Integer> createFlowPublisher(long elements) {
return new FlowPublisher((int) elements); // <-- how to t
浏览 1提问于2019-07-11得票数 0
2回答
一个简单的问题。假设我在下面的例子中实现了一个类。
class Subscriber
{
private Publisher publisher = new Publisher;
public Subscriber()
{
publisher.SomeEvent += new EventHandler(OnEventFired);
}
private void OnEventFired(object sender, EventArgs e)
{
}
}
在程序中的某个地方，我有一个方法，看起来像这样：
public
浏览 6提问于2011-07-13得票数 8
回答已采纳
目前，我有一个AWS EC2实例，它处理所有传入的http客户端请求。它分析每个请求，然后决定哪个后端工作者服务器应该处理请求，然后对所选服务器进行http调用。然后，后端服务器在处理请求时进行响应。然后，前端服务器将响应客户端。前端服务器实际上是一个负载均衡器。
现在我想使用Pub体系结构，而不是前端服务器，将请求推送到后端实例。前端服务器将进行一些基本处理，然后简单地将请求放入SNS队列，后端服务器应该处理的逻辑由后端服务器自己处理。
我的问题是，对于这个模型，让后端服务器通知前端服务器他们已经处理了请求，最好的方法是什么？以前，他们只是回复前端服务器发送的http请求，但现在没有直接请求
浏览 1提问于2018-04-26得票数 2
1回答
我们的团队最近在Android项目中采用了Kotlin Flow。 据我所知，Android中flows的一个缺点是它不是LifeCycleAware。这就是为什么我们一直使用LiveData将视图状态和数据链接到我们的视图。使用这两者的一种简单方法是让我们的Room查询返回流。这样，一旦我们的数据被转换，并且我们希望将其链接到视图，我们就可以使用asLiveData()函数。 现在我的问题是:总是转换为liveData的东西对我来说是不对的。这条路对吗？转换为LiveData会有什么影响吗？我们使用一个返回liveData版本的重复房间查询不是更好吗？
浏览 100提问于2021-09-22得票数 0
我正在寻找一种方法来管理来自多个出版商的信息流。
一个人为的例子是这样的。
我的每个发布者都在提交作业，这些作业可以分解为更小的组件，可以按任何顺序进行处理。每个组件都需要几秒钟的处理时间。
发布者A向包含1000个较小组件的队列提交作业。使用者X订阅队列，并开始处理队列消息并将结果发送到结果队列。
作业进行到一半时，发布者B提交了一个包含1000个组件的作业。
我希望Consumer X现在可以在来自Publisher A和Publisher B的消息之间交替，这样Publisher B就不需要在接收结果之前等待第一个作业完成，而Publisher A只是在Publisher B有了作业之后
浏览 16提问于2016-07-29得票数 0
1回答
所以我想知道一个问题。假设发布者在上午10:00创建了一个主题，并立即开始在该主题上发布消息。它通知消费者主题的创建，然后他们在上午10:01订阅主题这些消费者不会在上午10:00 - 10:01之间收到发布者发送的消息因此，如果消息传递提供程序发现此主题没有活动订阅者来接收任何消息，它是否应该立即丢弃这些消息。我们是否可以提供一种订阅机制，通过该机制，消费者可以指定他们希望从哪个点接收消息，(例如，从主题开始或2011年12/25上午10:00太平洋标准时间等)
浏览 0提问于2012-09-05得票数 1
我无法将文档保存到couchbase。
这是我的代码库。我在码头经营库奇基地。
我使用查询在DB中手动添加了一个文档。
repo.findAll()可以正常工作，但是repo.save(person)不能工作。
有人能帮我一下吗？
Person.java
@Document
@Getter
@Setter
@AllArgsConstructor
public class Person {
@Id
private String id;
@Field
@NotNull
private String firstName;
@Field
@
浏览 0提问于2018-03-10得票数 1
回答已采纳
刚刚开始学习NServiceBus，并试图理解它的概念。当它谈到队列时，我们是在谈论发布者和订阅者上的MSMQ吗？因此，如果我有一个应用程序生成了某个东西的列表(比如动物的名字)，那么它会将该列表转储到publisher的队列中。发布者每分钟轮询队列一次，如果队列中有内容，它将发布到订阅者的队列中进行进一步处理。这有意义吗？
谢谢。
浏览 1提问于2011-03-02得票数 2
回答已采纳
最近我读了一篇物联网的文章，他说，通过协议，客户端可以通过一对多和多对一的方式进行通信。在mqtt规范本身中提到了一对多，但我在多对一中几乎没有混淆。我所理解的是，许多客户端在同一主题上发布，并且这些不同的publishers.Then只有一个订阅者，订阅者(用户)应该如何识别消息属于哪个发布者？另一个疑问是，如果两个发布者使用相同的主题发布他们的数据而不知道对方，因为每个客户端都连接到互联网，订阅者将获得这两个数据。但这并不是订阅者expected.He想要的仅来自一个发布者的数据。在mqtt通信中有没有可能发生同样的情况？
浏览 1提问于2015-09-12得票数 8
这个问题已经在Akka Streams和Akka HTTP的旧版本中被问到了。这个答案的问题是，对于每个流，你需要为每个请求物化一个新的流，这并不便宜。
一年后，在最新版本中，将流与路由DSL集成的最佳方式是什么？
同样，我有一个想要添加到路由的复杂Flow，如果我将流映射到Flow[HttpRequest, HttpResponse, _]，我可以使用此流完成路由吗？有没有更好的方法或者是使用低级API的唯一方法？
如果是这样的话，做路由映射的最好方法是什么？
浏览 0提问于2017-05-23得票数 2
1回答
移动与ESP8266的连接
、、、、
我正在从事一个家庭自动化项目，我想将移动应用程序连接到ESP8266。
我可以发布从Mobile到AWS服务器的消息，AWS服务器是作为代理设置的。(工作)
我可以发布一条从AWS到Raspberry PI的消息。(工作)
我还可以发布从Raspberry到NodeMCU / ESP8266的Mqtt消息。(工作)
但是，当我试图把所有这些都整合在一起时，问题就会发生。当发布来自AWS的消息时，Raspberry会接收它，但它不会反映在NodeMCU / ESP8266上。
这是。
这就是我在我的ESP8266中使用的东西。当我发布从Raspberry PI，它是很好的工作
浏览 1提问于2016-08-22得票数 11
As事件网格定制主题是由发布者用来引发事件的http端点，而事件网格定制主题具有订阅者的webhook端点。如何通过事件网格自定义主题路由将响应从订阅服务器发送回发布者？
浏览 15提问于2020-08-01得票数 1
我正在尝试了解akka streams和处理web套接字的方法，但有些事情对我来说相当清楚。对于初学者来说，我正在尝试实现从某个客户端到服务器的单向通信，以及同一服务器和其他客户端之间的通信。
client1 -----> Server <------> client2
我正在查看提供的示例。生成的代码如下所示：
1)从控制器开始
class Test @Inject()(@Named("connManager") myConnectionsManager: ActorRef, cc: ControllerComponents)
浏览 0提问于2018-09-28得票数 0
虽然我看过了指南，但我找不到一种方法来做以下事情：
我们有n个出版商，
我们有百万订阅者。
每个订阅者订阅某种类型的消息(一个发布者可以发送一种以上的消息，多个pubs可以发出相同类型的消息)。
如何在0MQ中创建N到N(或N到1到N)模式？
浏览 2提问于2012-09-04得票数 8
2回答
我们使用Akka HTTP来使用akka streams API来处理我们的网络套接字连接。我们使用的是一个流，它将传入的消息传输到“连接参与者”。代码片段如下：
val connection = system.actorOf(ConnectionActor.props())
val in = Flow[Message]
.to(Sink.actorRef[Message](connection, WebSocketClosed))
val out = Source
.actorRef[Message](500, OverflowStrategy.fail)
.mapMater
浏览 0提问于2018-06-19得票数 1
回答已采纳
2回答
我使用一个参与者作为接收器，如下所示： import akka.Done
import akka.actor.{Actor, ActorLogging, Props}
import akka.http.scaladsl.model.StatusCodes
import akka.http.scaladsl.Http
import akka.http.scaladsl.model.ws.{Message, TextMessage, WebSocketRequest}
import akka.stream.{ActorMaterializer, Materializer}
import akka
浏览 19提问于2019-04-16得票数 0
根据我在MQTT协议中读到的内容，它似乎不支持告知已发布消息的发布者是谁。但是，MQTT订阅者有可能知道消息来自哪个发布者吗？
也许是“msg.publisher”的变通方法？
#!/usr/bin/env python
import mosquitto
def on_message(mosq, obj, msg):
print "Publisher: %s, Topic: %s, "Msg: %s" % (msg.publisher, msg.topic, msg.payload)
cli = mosquitto.Mosquitto()
cli.on_me
浏览 22提问于2015-09-16得票数 4
1回答
阅读akka的文档，我不太清楚消息的顺序，以及是否可以强制执行。让我用我为聊天服务器编写的一小部分代码来设置我问题的上下文。
def flowShape(user: User) = GraphDSL
.create(Source.actorRef[ChatMessage](bufferSize = 5, OverflowStrategy.fail)) {
implicit builder =>
implicit chatSource =>
import GraphDSL.Implicits._
val messageFromOut
浏览 4提问于2017-11-11得票数 0
回答已采纳
1回答
我已经看到了使用Akka Streams来避免背压的例子，但是我想知道如果我需要在下游获得自定义的acks，这是如何工作的？
在我的特定情况下，链中有几个处理消息的flows，当消息到达某个flow时，我想将一个ack传递回源，以便发出信号，表明我现在已经“取得了”正在处理的资源的所有权。
这在Akka Streams中是如何工作的？
浏览 0提问于2017-04-03得票数 1
我想创建一个向多个用户广播流的android应用程序。我正在为广播者和订阅者使用发布者令牌，因为我希望看到来自所有订阅了主流的订阅者的流。因为所有的订阅者都在发布流simultaneously.The错误，我遇到的错误是，当一个新订阅者加入会话时，新订阅者订阅另一个订阅者(用户)的steam，而不是我希望他们订阅的主流。你能告诉我，我如何确保会话中的每个人都订阅来自特定人的特定流？
浏览 13提问于2018-02-07得票数 0
1回答
订阅客户端未收到消息MQTT
、、、、
我正在研究MQTT协议，我尝试发布和订阅两个不同的Java应用程序。
我的第一个应用程序是“发布”。我在MQTT服务器上发布了一条消息。
我的第二个应用程序是“订阅”。我订阅了一个主题，并尝试接收消息。但是我从来没有收到过这条消息。
当我运行这两个应用程序时，我从“订阅”应用程序开始，然后运行“发布”应用程序。当“发布”应用程序开始时，我失去了与“订阅”应用程序的连接，并且我无法接收我的消息。
在"Subscribe“应用程序中，client.setCallback(this)从不调用我的方法messageArrived()。(请参阅下面的代码)。
下面是我的两个代码应用程序：
发布应
浏览 3提问于2016-05-26得票数 0
我在Scala项目中使用Akka Streams API，在Intellij中使用SBT插件。我有一个员工池流，如下所述：。这是我的代码：
package streams
import akka.NotUsed
import akka.stream.scaladsl.{Balance, Flow, GraphDSL, Merge}
import akka.stream.{FlowShape, Graph}
object WorkerPoolFlow {
def apply[In, Out](
worker: Flow[In, Out, Any],
workerCount
浏览 2提问于2017-11-18得票数 2
回答已采纳
1回答
我已经成功地将MQTT连接到我的代理发布者和订阅者都工作得很好，但存在一个问题，我的订阅者没有主动接收消息，当我重新启动订阅者应用程序时，函数mqtt_publishedRecieved仅触发一次，并且一次只接收一条消息。为了获得另一条消息，我必须再次重启我的应用程序。据我所知，这是因为我在启动时调用了我的类的config方法。因此，它只在启动时检查订阅的主题。但我真的希望我的subscriber类中的功能在接收方发布消息后立即接收消息，这意味着两者应该并行工作。我的要求是，如果订阅者已连接，则为=>，在发布者触发时立即接收消息。=>如果订阅者断开连接，则消息应排队，稍后当订阅者
浏览 56提问于2021-11-07得票数 1
回答已采纳
1回答
有没有办法认为MQXR是可伸缩的？
如果订阅者不知道发布者是谁-如何确保负载平衡的发布者可以发布到负载平衡的订阅者？
如果我的术语都错了，我很高兴被纠正。
浏览 6提问于2021-11-11得票数 0
我想知道zmq是否已经解决了以下问题(或者)位于zmq之上的应用程序需要处理它。
1)向所有订阅者发布数据的中央发布者。这些数据本质上是静态的，类似于配置。可以在任何时间点修改数据。
2)多个订阅者订阅来自该发布者的消息。发布者可以在任何时间点加入。
3)如果数据发生变化，发布者应该只将差异发布到现有的订阅者。
4)如果订阅者稍后加入，发布者应将所有数据(当前配置)发布给新订阅者。
Zeromq指南建议遵循以下方法来解决慢Joiner综合征，但这并不能解决上述问题。
浏览 0提问于2013-05-04得票数 3
回答已采纳
在我的消息发布者配置中，我
<MsmqTransportConfig
InputQueue="EnformMessages"
ErrorQueue="error"
NumberOfWorkerThreads="1"
MaxRetries="5"
/>
<UnicastBusConfig ForwardReceivedMessagesTo="testqueue@cgy1-web01">
<MessageEndpointMappi
浏览 0提问于2010-03-05得票数 1
我正在尝试创建一个事件处理程序，它将遍历相同类型的委托列表，并在需要时调用它们。我最终想要的是一个“主”处理程序，它控制调用哪些子处理程序(这是用于实现默认事件处理的嵌套状态机，因此如果在嵌套级别中处理事件，则阻止它向上传播到更高级别)。 我最简单的解决方案是只订阅“最深”的事件处理程序。但这并不是我真正想要的行为，我希望该处理程序的主体表明它是否真的“处理”了事件，如果它没有“处理”事件，则调用下一个级别。 所以我的第二种方法是事件只有一个实际的订阅者。这会依次遍历委托列表，并使用修改后的方法签名调用它们，如果它处理了事件，该签名将返回一个布尔值。 我的.NET事件定义：
浏览 27提问于2019-04-06得票数 0
我正在使用Xcode6和swift制作一个支持视频聊天的应用程序--使用OpenTok。我有两个场景“菜单”和“视频”。从我使用的菜单中：
self.performSegueWithIdentifier("StreamScene", sender: self)
在按钮内转到“视频”视图控制器。当我第一次加载我的应用程序时，一切都很完美，但是当我从“视频”导航到“菜单”，然后再次从“菜单”导航到“视频”时，我看不到我的会话的订阅者，但当我关闭我的应用程序，但仍然让视频会话运行，并再次启动我的应用程序，并导航到“视频”视窗控制器它工作-这是因为我的ViewControler可能不能
浏览 0提问于2015-08-12得票数 0
我的团队正在寻找将系统的各个组件分离为解耦服务的方法。我们想要实现的是一种事件驱动的模式，在该模式中，当事件在其他系统中发生时，服务订阅以接收发送的信息。
由于我们已经将WCF用于请求-回复消息传递，因此我们希望继续将其用于此特定模式。理想情况下，消息将通过MSMQ进行管理，以使我们能够使用可靠的消息传递，以便在发生服务故障时提供容错能力。
我们并不是在寻找任何复杂的东西，比如跨服务边界的事务支持。实际上，我们只需要一个简单的基于订阅的消息分发系统。有没有什么简单的框架或服务可以帮助我们实现这种模式？
浏览 0提问于2011-09-08得票数 1
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
即时通信 IM活动推荐
腾讯云开发者
