Oracle AQ与.NET Core：消息可用通知事件失效问题求助

问题概述 #

需实现Oracle AQ持续读取的Worker Service，预期流程：

1. 打开数据库连接
2. 创建队列对象
3. 绑定MessageAvailable事件
4. 调用Listen()启动监听
5. 事件处理中处理消息并提交（队列层面提交，非事务提交）
   需求支持10-15个并行消费者，但当前OnMessageAvailable事件从未触发，使用Oracle.ManagedDataAccess.Core 23.6.1版本。

代码中的关键问题 #

• 同步Dequeue阻塞监听流程：代码中绑定事件后立即调用queue.Dequeue()，该操作会同步尝试获取消息，若队列无消息会直接抛出异常或返回null，打断后续监听逻辑的执行，且与异步监听模式冲突。
• DequeueOptions.Wait配置错误：Wait=0表示不等待立即返回，监听模式下需设置为-1（无限等待）或大于0的超时值，否则Listen无法持续等待新消息，导致监听直接结束。
• Payload类型转换错误：队列使用UDT类型BR_AQADM.TJSONRESPONSE，message.Payload实际是TJsonResponse实例，代码中强行转为byte[]会得到null，引发后续处理错误。
• Listen方法调用时机与参数：Listen方法需在事件绑定完成后调用，且参数需与队列配置的NotificationConsumers一致，同时需确保数据库队列已启用通知功能。

修复后的代码示例 #

using Oracle.ManagedDataAccess.Client;
using Oracle.ManagedDataAccess.Types;
using System;
using System.Threading.Tasks;

namespace OracleAQReader;

class Program
{
    static void Main()
    {
        // 启动12个并行消费者，可根据需求调整数量
        var consumerTasks = new Task[12];
        for (int i = 0; i < consumerTasks.Length; i++)
        {
            consumerTasks[i] = Task.Run(StartConsumer);
        }

        Console.WriteLine("所有消费者已启动，按Enter键退出...");
        Console.ReadLine();
    }

    static void StartConsumer()
    {
        string connectionString = "你的数据库连接字符串";
        // 每个消费者使用独立连接，避免线程安全问题
        using (OracleConnection connection = new OracleConnection(connectionString))
        {
            connection.Open();

            OracleAQQueue queue = new OracleAQQueue("BR_AQADM.AQ_TRANSACTIONNOTIFICATION", connection, OracleAQMessageType.Udt, "BR_AQADM.TJSONRESPONSE")
            {
                NotificationConsumers = ["TRANSACTIONSSUBSCRIBER"],
                DequeueOptions = new OracleAQDequeueOptions
                {
                    Wait = -1, // 无限等待新消息
                    ConsumerName = "TRANSACTIONSSUBSCRIBER",
                    NavigationMode = OracleAQNavigationMode.NextMessage,
                    DequeueMode = OracleAQDequeueMode.Remove
                }
            };

            Console.WriteLine($"消费者 {Task.CurrentId} 已连接，连接状态: {queue.Connection.State}");

            queue.MessageAvailable += OnMessageAvailable;

            // 启动监听，该方法会阻塞当前线程直至停止监听
            queue.Listen(new[] { "TRANSACTIONSSUBSCRIBER" });
        }
    }

    static void OnMessageAvailable(object sender, OracleAQMessageAvailableEventArgs e)
    {
        OracleAQQueue queue = (OracleAQQueue)sender;
        try
        {
            OracleAQMessage message = queue.Dequeue();
            if (message.Payload is TJsonResponse response)
            {
                string messageContent = response.Response;
                Console.WriteLine($"消费者 {Task.CurrentId} 收到消息: {messageContent}");

                // 队列层面提交消息（非事务提交）
                queue.CommitMessage(message);
            }
        }
        catch (Exception ex)
        {
            Console.WriteLine($"消息处理出错: {ex.Message}");
        }
    }
}

public class TJsonResponse : IOracleCustomType, INullable
{
    [OracleObjectMapping("RESPONSE")]
    public string Response { get; set; }

    public bool IsNull { get; set; }

    public IOracleCustomType CreateObject()
    {
        return new TJsonResponse();
    }

    public void FromCustomObject(OracleConnection con, object udt)
    {
        OracleUdt.SetValue(con, udt, "RESPONSE", Response);
    }

    public void ToCustomObject(OracleConnection con, object udt)
    {
        Response = (string)OracleUdt.GetValue(con, udt, "RESPONSE");
    }
}

[OracleCustomTypeMappingAttribute("BR_AQADM.TJSONRESPONSE")]
public class TJsonResponseTypeFactory : IOracleCustomTypeFactory
{
    public IOracleCustomType CreateObject()
    {
        return new TJsonResponse();
    }
}

额外注意事项 #

• 并行消费者实现：每个消费者必须使用独立的OracleConnection和OracleAQQueue实例，Oracle.ManagedDataAccess.Core的队列对象并非线程安全，共享实例会引发竞争问题。
• 队列通知配置：确保数据库中的队列已启用通知，可通过SQL查询验证：SELECT name, enabled FROM user_queues WHERE name = 'AQ_TRANSACTIONNOTIFICATION';，若enabled为N，需执行ALTER QUEUE BR_AQADM.AQ_TRANSACTIONNOTIFICATION ENABLE;启用。
• 事务与提交：CommitMessage是队列层面的提交，不会影响数据库事务，若需事务控制需手动管理OracleTransaction。

内容的提问来源于stack exchange，提问作者Leonardo