Oracle 11g AQ订阅代理lasernet脱队挂起问题求助

我之前碰到过几乎一模一样的Oracle 11g AQ订阅代理挂起问题，结合你的代码和场景，分享几个亲测有效的排查和解决方向：

1. 先排查代理状态与异常消息阻塞 #

Oracle AQ的订阅代理遇到无法处理的消息时，很容易陷入“假死”状态。先通过PL/SQL确认代理和队列的状态：

• 检查订阅代理的配置是否正常：

SELECT q.name AS queue_name, s.subscriber_name, s.rule, s.delivery_mode
FROM dba_queue_subscribers s
JOIN dba_queues q ON s.queue_owner = q.owner AND s.queue_name = q.name
WHERE q.name = 'XALOUTPUT_Q' AND s.subscriber_name = 'LASERNET';

• 查看队列中是否有异常状态的消息（这类消息会阻塞代理消费）：

SELECT COUNT(*) FROM XAL_SUPERVISOR.AQ$XALOUTPUT_Q_TABLE WHERE STATE = 3; -- STATE=3代表异常消息

如果存在异常消息，建议将其转移到专门的异常队列（或者直接清理），之后重启代理消费，大概率能恢复正常。

2. 修复.NET代码中的递归与资源泄漏问题 #

看你的Listen()方法，finally和异常块里都递归调用了Listen()，这种写法很容易导致栈溢出和数据库连接/队列对象泄漏，进而引发Oracle端的代理会话异常挂起。

推荐改成循环+using自动释放资源的写法：

public bool Listen()
{
    while (true) // 用循环替代递归，避免栈溢出
    {
        // using自动释放连接资源
        using (var conListen = new OracleConnection(_constr))
        {
            try
            {
                conListen.Open();
                XALOUTPUT_T queueItem = null;
                var threadNumber = CountThread(SUBSCRIBER_1);
                Console.WriteLine("TestListen()");

                // using自动释放队列对象
                using (OracleAQQueue queueListen = new OracleAQQueue("XALOUTPUT_Q", conListen))
                {
                    string subscriber = "lasernet";
                    var listenConsumers = new string[2] { subscriber, "vau_mail" };

                    queueListen.MessageType = OracleAQMessageType.Udt;
                    queueListen.UdtTypeName = "XALOUTPUT_T";
                    queueListen.DequeueOptions.ConsumerName = subscriber.ToUpper();
                    queueListen.DequeueOptions.DeliveryMode = OracleAQMessageDeliveryMode.PersistentOrBuffered;
                    queueListen.DequeueOptions.Visibility = OracleAQVisibilityMode.Immediate;
                    queueListen.DequeueOptions.Wait = 5;

                    var consumer = queueListen.Listen(listenConsumers, 5);
                    if (consumer.ToUpper() == "LASERNET")
                    {
                        Console.WriteLine("[Listen Thread] Listen returned... Dequeuing...");
                        OracleAQMessage deqMsg = queueListen.Dequeue(queueListen.DequeueOptions);
                        XALOUTPUT_T result = (XALOUTPUT_T)deqMsg.Payload;

                        try
                        {
                            if (deqMsg != null && deqMsg.Payload is XALOUTPUT_T queuePayload)
                                queueItem = queuePayload;

                            var operation = SendDataToLasernet(queueItem, threadNumber).GetAwaiter().GetResult();
                        }
                        catch (Exception e)
                        {
                            Console.WriteLine($"Processing Error: {e.Message}");
                            // 关键：把处理失败的消息转移到异常队列，避免阻塞代理
                            deqMsg.EnqueueOptions.Visibility = OracleAQVisibilityMode.Immediate;
                            deqMsg.Enqueue("XAL_SUPERVISOR.XALOUTPUT_Q_EXCEPTION"); // 需提前创建异常队列
                        }
                    }
                    else
                    {
                        Console.WriteLine("Consumer not found...");
                    }
                }
            }
            catch (Exception e)
            {
                Console.WriteLine(".NET - Error: {0}", e.Message);
                // 异常后暂停几秒再重试，避免频繁创建连接
                Thread.Sleep(5000);
            }
        }
    }
}

核心改进点：

• 用while(true)循环替代递归，防止栈溢出
• 用using语句自动释放OracleConnection和OracleAQQueue，彻底避免资源泄漏
• 处理消息异常时，将失败消息转移到异常队列，不让代理反复尝试处理无效消息

3. 验证订阅规则的匹配逻辑 #

你的订阅规则是tab.user_data.beskedtype = 'lasernet'，要确保队列中的消息确实符合这个规则：

• 测试插入一条明确匹配规则的消息，看代理是否能正常消费：

DECLARE
  msg XALOUTPUT_T;
BEGIN
  -- 构造一个beskedtype='lasernet'的测试消息
  msg := XALOUTPUT_T(beskedtype => 'lasernet', /* 其他字段按需填充 */);
  DBMS_AQ.ENQUEUE(
    queue_name => 'XAL_SUPERVISOR.XALOUTPUT_Q',
    enqueue_options => DBMS_AQ.enqueue_options_t(),
    message_properties => DBMS_AQ.message_properties_t(),
    payload => msg,
    msgid => NULL);
  COMMIT;
END;
/

如果测试消息能被正常消费，说明之前的问题可能是消息不匹配导致的“看似代理停止工作”；如果还是不行，继续往下排查。

4. 检查Oracle AQ后台进程状态 #

Oracle 11g的AQ依赖AQ_TM_PROCESSES后台进程，确保它处于正常运行状态：

-- 查看当前AQ_TM_PROCESSES配置
SELECT value FROM v$parameter WHERE name = 'aq_tm_processes';

-- 如果值为0，设置为10（根据业务量调整）
ALTER SYSTEM SET aq_tm_processes = 10 SCOPE=BOTH;

同时检查队列本身的状态：

SELECT name, queue_state FROM dba_queues WHERE name = 'XALOUTPUT_Q';
-- 正常状态应为ENABLED

5. 临时兜底方案：自动重建代理 #

如果以上方法都无法彻底解决，可以写一个定时任务（Oracle Job或Windows服务定时线程），定期检查代理状态，发现异常自动重建：

DECLARE
  v_exists NUMBER;
BEGIN
  -- 检查代理是否存在
  SELECT COUNT(*) INTO v_exists
  FROM dba_queue_subscribers
  WHERE queue_owner = 'XAL_SUPERVISOR' AND queue_name = 'XALOUTPUT_Q' AND subscriber_name = 'LASERNET';
  
  IF v_exists = 0 THEN
    -- 重建代理
    DECLARE
      subscriber sys.aq$_agent;
    BEGIN
      subscriber := sys.aq$_agent ('lasernet', 'XALOUTPUT_Q', 0);
      DBMS_AQADM.ADD_SUBSCRIBER(
        queue_name => 'XAL_SUPERVISOR.XALOUTPUT_Q',
        subscriber => subscriber,
        RULE => 'tab.user_data.beskedtype = ''lasernet''' ,
        TRANSFORMATION => null,
        QUEUE_TO_QUEUE => false,
        DELIVERY_MODE => 3);
      COMMIT;
    END;
  END IF;
END;
/

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