- html - 出于某种原因,IE8 对我的 Sass 文件中继承的 html5 CSS 不友好?
- JMeter 在响应断言中使用 span 标签的问题
- html - 在 :hover and :active? 上具有不同效果的 CSS 动画
- html - 相对于居中的 html 内容固定的 CSS 重复背景?
我有一个应用程序,我可以在流程的一部分中以 JSON 格式将消息写入 Azure 服务总线队列。我有一个下游进程,我希望将该消息从队列中弹出,将 json 转换为对象,然后处理该对象。
我将消息推送到队列中没有问题,但我找不到任何一次从队列中弹出消息或循环弹出消息的示例。我在 Microsoft 或 Github 上看到的每个示例都是一个控制台应用程序(在 Web 应用程序中无用),它设置某种监听器来捕获队列中的所有消息并写入控制台消息。我没有找到弹出消息然后对数据进行一些处理的示例。有没有人有关于如何从队列中弹出消息然后处理它或调用另一个方法来对消息中的数据执行某些操作的示例?
更新:我使用了下面由 Guru Pasupathy 提供的 WindowsAzure.ServiceBus 示例,并使用 Azure Service Bus Queues: How To Read Individual Messages From A Queue 中的以下代码片段结束。从 BrokeredMessage 对象获取消息文本:
Stream stream = message.GetBody<Stream>();
StreamReader reader = new StreamReader(stream);
string messageBody = reader.ReadToEnd();
然后我可以获取 messageBody 并将嵌入的 JSON 反序列化为 POCO 对象,然后我就可以了!现在,我可以在应用程序中更有效地使用队列来执行各种任务。
最佳答案
您可以使用 Peek Lock 接收模式从队列中获取消息,处理它,然后您可以根据您的业务逻辑选择放弃它或完成它。
如果您使用WindowsAzure.ServiceBus nuget包发送/接收消息,则可以使用以下方法基于循环或单次调用逐一使用队列中的消息,而无需使用监听器。
public void Receive()
{
QueueClient myQueueClient = QueueClient.CreateFromConnectionString("<connectionString>;<queueName>", ReceiveMode.PeekLock);
int someCount = 2; //some random value for testing
try
{
for (int i = 0; i < someCount; i++)
{
BrokeredMessage message = myQueueClient.Receive();
Console.WriteLine("The message is " + message);
message.Complete();
}
}
catch(Exception e)
{
//Handle your expection
}
}
如果您使用的是 Microsoft.Azure.Service nuget 包,那么我找不到在不使用监听器的情况下仅读取单个消息的直接方法。我看到监听器将继续轮询和处理队列,直到没有更多消息为止。
如果您的要求是停止轮询并继续获取所有可用消息,那么作为一种解决方法,您可以在读取一条消息后关闭 QueueClient 实例,并在进程准备好接受下一条消息时打开 qc 并注册处理程序.
public async Task ProcessMessagesAsync(Message message, CancellationToken token)
{
Console.WriteLine($"Received message: {Encoding.UTF8.GetString(message.Body)}");
BinaryFormatter bf = new BinaryFormatter();
using (MemoryStream ms = new MemoryStream(message.Body))
{
Payload payload = (Payload) bf.Deserialize(ms);
//Based on your needs you may have a condition here based on which you could Abandon or Complete the mesage
await qc.CompleteAsync(message.SystemProperties.LockToken);
Console.WriteLine("Completed the message --> " + payload.Message + " -- Id --> " + payload.Id);
//await qc.AbandonAsync(message.SystemProperties.LockToken);
//Console.WriteLine("Abandon the message --> " + payload.Message + " -- Id --> " + payload.Id);
qc.CloseAsync(); //If you close the QueueClient instance here, no more messages will be picked up from queue.
}
}
以上示例基于 https://learn.microsoft.com/en-us/azure/service-bus-messaging/service-bus-dotnet-get-started-with-queues#receive-messages-from-the-queue ,我刚刚在末尾添加了 qc.CloseAsync() 调用。如果没有这一行,监听器将继续处理,直到队列中不再有消息为止。我不确定是否有更好的方法来实现这一点,但想到分享。
希望这有帮助
编辑-
发送消息时,如果您使用自定义类型,可以使用以下内容
BrokeredMessage message = new BrokeredMessage(new Payload() { Id = 4332, Message = "WindowsAzure package" });
myQueueClient.Send(message);
在接收时,您应该使用 GetBody,如下所示
BrokeredMessage message = myQueueClient.Receive();
var incoming = message.GetBody<Payload>();
Console.WriteLine("The message is " + incoming.Id + " and " + incoming.Message);
message.Complete();
以下是供您引用的自定义对象
[Serializable]
public class Payload
{
public int Id { get; set; }
public string Message { get; set; }
}
同样适用于字符串
接收时您可以使用
var incoming = message.GetBody<string>();
发送时您可以发送为
BrokeredMessage message = new BrokeredMessage("WindowsAzure package" );
您可以通过以下链接获取有关不同内容格式的更多详细信息 https://abhishekrlal.com/2012/03/30/formatting-the-content-for-service-bus-messages/
关于Azure服务总线队列: How To Read Individual Messages From A Queue,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/61977591/
我有一个阅读器,其中包含有关 51*51 网格的信息,其中网格上的每个点都由 f32 表示。 .我想将这些数据读入一个向量,以便我可以轻松处理它: pub fn from_reader(reader:
我重新启动了 SQL Server 2005 并运行了统计 IO 的查询。 我得到了这些结果:表“xxx”。扫描计数 1,逻辑读取 789,物理读取 3,预读读取 794,... 预读读取数是读取并放
在 CLHS 中,我为 :read-only x 读到:“当 x 为真时,这指定不能更改此插槽;它将始终包含构造时提供的值。” 我可以做到这一点(CCL、SBCL): CL-USER> (defstr
让我们考虑一下这句话(Total Store Ordering): reads are ordered before reads, writes before writes, and reads be
我正在开发一个 SMTP 库,它使用缓冲读取器通过网络读取行。 我想要一种安全的方式来从网络读取数据,而不依赖于 Rust 内部机制来确保代码按预期工作。具体来说,我想知道 Read trait 是否
我不清楚所有这些读取字符串函数之间的关系。嗯,很明显clojure.core/read-string可以读取 pr[n] 输出的任何序列化字符串甚至 print-dup .也很清楚clojure.ed
所以我做了这个功能,就像倒计时一样。我想在倒计时减少时读取命令。我的大问题是让 read() 在倒计时减少时等待输入。如您所见,我尝试使用 select() 但在第一个 printf 之后("time
这是我vue3+echart5 遇到的报错:Cannot read properties of undefined (reading ‘type‘) 这个问题需要搞清楚两个关键方法: toRaw: 作
下图中,左边是C代码,右边是未优化的LLVM IR形式。 The Figure 在 IR 上运行 MemoryDependenceAnalysis 可查找内存依赖性。原始代码及其 IR 等效代码中
这个问题在这里已经有了答案: Read values into a shell variable from a pipe (17 个答案) 关闭 3 年前。 我一直在尝试像这样从程序输出中读取环境变
当我输入相同的整数时,如何将整数转换为与使用 read(0,buff,nbytes) 获得的缓冲区相同的值/编码字符?我正在尝试编写类似 read() 的东西,但用整数数据代替读取到缓冲区的文件描述符
This question already has answers here: Closed 2 years ago. Read input in bash inside a while loop (
我正在尝试处理来自 MySQL 数据库的一些数据(主要是 double 值)。我收到此错误消息: Invalid attempt to access a field before calling Re
我正在制作一个简单的 TCP/IP 套接字应用 这样做有什么不同: DataInputStream in = new DataInputStream(clientSocket.getInputStre
我操作API服务器。 手机APP访问API服务器时,有时会出现该异常。 我尝试在测试服务器上进行测试,但无法重现。(我改变了apache和tomcat的连接时间。) 有什么问题?? 我该如何解决这个问
我在段落末尾使用“阅读更多”只是为了提醒像P.T.O一样的用户 为什么会有问题? 最佳答案 您必须明白,许多屏幕阅读器用户不会等到整个页面都读给他们听。他们使用键盘快捷键在页面中导航。 JAWS(可以
我已将我的 Angular 应用程序从 12 版本升级到 13 版本。我在单元测试运行期间开始遇到此错误。 Chrome Headless 94.0.4606.61 (Windows 10) AppC
我正在尝试为以下组件编写一个。我正在使用 queryParams 然后使用 switchmap 来调用服务。这是 url 的样子: http://localhost:4200/test-fee/det
我的代码有什么问题? Uncaught TypeError: Cannot read properties of undefined (reading 'remove') 和 Uncaught Typ
我在我的 React 应用程序中遇到了这个问题。 类型错误:无法读取未定义的属性(读取“requestContent”) 我在我的应用程序中使用 commercejs。代码指向 isEmpty=!ca
我是一名优秀的程序员,十分优秀!