2

我正在处理从一个数据库到另一个数据库的 AQ 传播，但是当我安排传播并将第一条消息排队到 LOCAL AQ 表时，我在 DBA_QUEUE_SCHEDULES.LAST_ERROR_MSG 中遇到错误，“ORA-25215：user_data 类型和队列类型不匹配”。请注意，AQ 表中使用的两种对象类型是相同的，出于测试目的，我使用的是这个：

create or replace type LOCAL_OBJ_MSG as object(
    test varchar2(4000))
/

并且两个 AQ 表也是相同的，我使用相同的脚本创建它们并且只更改了名称，一个 LOCAL 另一个 REMOTE。LOCAL AQ 表在 LOCAL 数据库中，REMOTE 表在 SCHEMA_NAME.REMOTE 数据库中。

这是我用来创建 AQ 表的脚本：

declare
  QUEUE_NOT_FOUND exception;
  PRAGMA EXCEPTION_INIT(QUEUE_NOT_FOUND, -24010);
begin
  dbms_aqadm.stop_queue(
    queue_name => 'REMOTE_iTEST');
exception
  when QUEUE_NOT_FOUND then
    null;
  when others then
    raise;
end;
/

declare
  QUEUE_NOT_FOUND exception;
  PRAGMA EXCEPTION_INIT(QUEUE_NOT_FOUND, -24010);
begin
  dbms_aqadm.drop_queue('REMOTE_iTEST');
exception
  when QUEUE_NOT_FOUND then
    null;
  when others then
    raise;
end;
/

declare
  QUEUE_NOT_FOUND exception;
  PRAGMA EXCEPTION_INIT(QUEUE_NOT_FOUND, -24002);
begin
  dbms_aqadm.drop_queue_table('REMOTE_iTEST', force => true);
exception
  when QUEUE_NOT_FOUND then
    null;
  when others then
    raise;
end;
/

declare
  QUEUE_NOT_FOUND exception;
  PRAGMA EXCEPTION_INIT(QUEUE_NOT_FOUND, -24010);
begin
  dbms_aqadm.stop_queue(
    queue_name => 'REMOTE_oTEST');
exception
  when QUEUE_NOT_FOUND then
    null;
  when others then
    raise;
end;
/

declare
  QUEUE_NOT_FOUND exception;
  PRAGMA EXCEPTION_INIT(QUEUE_NOT_FOUND, -24010);
begin
  dbms_aqadm.drop_queue('REMOTE_oTEST');
exception
  when QUEUE_NOT_FOUND then
    null;
  when others then
    raise;
end;
/

declare
  QUEUE_NOT_FOUND exception;
  PRAGMA EXCEPTION_INIT(QUEUE_NOT_FOUND, -24002);
begin
  dbms_aqadm.drop_queue_table('REMOTE_oTEST', force => true);
exception
  when QUEUE_NOT_FOUND then
    null;
  when others then
    raise;
end;
/

declare
  l_exist number;
  cursor c_index is
    select 1 from user_objects u where u.OBJECT_NAME = upper('LOCAL_OBJ_MSG') and u.OBJECT_TYPE = 'TYPE';
begin
  open c_index;
  fetch c_index into l_exist;
  close c_index;

  if l_exist = 1 then
    execute immediate 'drop type LOCAL_OBJ_MSG';
  end if;
end;
/

create or replace type LOCAL_OBJ_MSG as object(
    test varchar2(4000))    
/

begin
  dbms_aqadm.create_queue_table (
    queue_table => 'REMOTE_iTEST',
    queue_payload_type => 'LOCAL_OBJ_MSG',
    storage_clause => 'pctfree 5 pctused 90 tablespace SEPA_INTEG_AQ',
    message_grouping => DBMS_AQADM.NONE,
    sort_list => 'ENQ_TIME',
    multiple_consumers => true,
    comment => 'Incoming TEST message table.');
end;
/

begin
  dbms_aqadm.create_queue (
    queue_table    => 'REMOTE_iTEST',
    queue_name     => 'REMOTE_iTEST',
    queue_type => sys.dbms_aqadm.normal_queue,
    retention_time => sys.dbms_aqadm.INFINITE,
    comment => 'Incoming TEST messages.',
    max_retries => 5);
end;
/

begin
  dbms_aqadm.start_queue(
    queue_name  => 'REMOTE_iTEST',
    dequeue     => true,
    enqueue     => true);
end;
/
begin
  dbms_aqadm.create_queue_table (
    queue_table => 'REMOTE_oTEST',
    queue_payload_type => 'LOCAL_OBJ_MSG',
    storage_clause => 'pctfree 5 pctused 90 tablespace SEPA_INTEG_AQ',
    message_grouping => DBMS_AQADM.NONE,
    sort_list => 'ENQ_TIME',
    multiple_consumers => true,
    comment => 'Outgoing TEST message table.');
end;
/

begin
  dbms_aqadm.create_queue (
    queue_table    => 'REMOTE_oTEST',
    queue_name     => 'REMOTE_oTEST',
    queue_type => sys.dbms_aqadm.normal_queue,
    retention_time => sys.dbms_aqadm.INFINITE,
    comment => 'Outgoing TEST messages.',
    max_retries => 5);
end;
/

begin
  dbms_aqadm.start_queue(
    queue_name  => 'REMOTE_oTEST',
    dequeue     => TRUE,
    enqueue     => TRUE);
end;
/

以下是用于创建订阅者、出列程序等的脚本：

本地数据库：

begin
  -- Add the remote subscriber.
  dbms_aqadm.add_subscriber(queue_name     => 'LOCAL_oTEST',
                            subscriber     => sys.aq$_agent(name     => 'LOCAL_oTEST_subscriber',
                                                            address  => 'SCHEMA_NAME.REMOTE_oTEST@DB_LINK_NAME',
                                                            protocol => 0),
                            queue_to_queue => true);
  -- Start the propagation of messages.
  dbms_aqadm.schedule_propagation(queue_name        => 'LOCAL_oTEST',
                                  latency           => 0,
                                  destination       => 'DB_LINK_NAME',
                                  destination_queue => 'SCHEMA_NAME.REMOTE_oTEST');
end;
/

远程数据库：

-- Create a table to store the messages received.
create table sepa_omsg_aq_demo
  (received timestamp default systimestamp,
   message LOCAL_OBJ_MSG);

-- Create a callback procedure that dequeues the received message and saves it
create or replace
procedure REMOTE_CALLBACK_TEST
  (
    context raw,
    reginfo sys.aq$_reg_info,
    descr sys.aq$_descriptor,
    payload raw,
    payloadl number
  )
as
  r_dequeue_options dbms_aq.dequeue_options_t;
  r_message_properties dbms_aq.message_properties_t;
  v_message_handle raw(26);
  o_payload LOCAL_OBJ_MSG;
begin
  r_dequeue_options.msgid         := descr.msg_id;
  r_dequeue_options.consumer_name := descr.consumer_name;
  dbms_aq.dequeue(queue_name => descr.queue_name, 
                  dequeue_options => r_dequeue_options, 
                  message_properties => r_message_properties, 
                  payload => o_payload, 
                  msgid => v_message_handle);
  insert into sepa_omsg_aq_demo 
    (message) 
    values (o_payload);
  commit;
exception
  when others then
    rollback;
end;
/

-- Register the procedure for dequeuing the messages received.
-- I'd like to point out that the subscriber is the one defined for the local database
begin
  dbms_aq.register (
     sys.aq$_reg_info_list(
        sys.aq$_reg_info('REMOTE_oTEST:LOCAL_oTEST_subscriber',
                         dbms_aq.namespace_aq,
                         'plsql://REMOTE_CALLBACK_TEST',
                         hextoraw('FF'))
                        ), 
        1);
end;
/

将消息排队到 LOCAL AQ 表的脚本：

declare
  enq_msgid raw(16);
  eopt      dbms_aq.enqueue_options_t;
  mprop     dbms_aq.message_properties_t;

  message local_obj_msg;
begin
  message := local_obj_msg('a');
  dbms_aq.enqueue(queue_name         => 'LOCAL_oTEST',
                  enqueue_options    => eopt,
                  message_properties => mprop,
                  payload            => message,
                  msgid              => enq_msgid);
  commit;
end;
/

另请注意，在 SYS.AQ$_MESSAGE_TYPES 表（LOCAL db）中，已验证状态 = 'F' 用于创建传播，并且：

declare
  rc binary_integer;
begin
  dbms_aqadm.verify_queue_types(src_queue_name  => 'local_otest',
                                dest_queue_name => 'schema_name.remote_otest',
                                rc              => rc,
                                destination     => 'db_link_name');
  dbms_output.put_line('Compatible: ' || rc);
end;
/

返回 0，这意味着表类型不兼容。在我进行了一些挖掘之后，我发现如果 LOCAL 和 REMOTE 数据库 NLS_LENGTH_SEMANTICS 不同，则无法进行远程传播，但是在这种情况下并非如此。我已经检查过了。任何想法是我做错了什么，或者我怎样才能找到这两个表之间的不同之处以及如何解决它？或者它可能是某些数据库参数值之间的差异？

Oracle 数据库 11g 版本 11.2.0.3.0

4

2 回答 2

1

要检查的一个区域是每个数据库上的 NLS_LENGHT_SEMANTICS 参数值。

NLS_LENGTH_SEMANTICS允许您以字符而不是字节来指定列数据类型的长度。

您的 PL/SQL 类型创建脚本不会在 VARCHAR2 长度内隐式使用 BYTE 或 CHAR 转换。这将导致类型的生成继承为NLS_LENGTH_SEMANTICS设置的值。

确保检查您的两个数据库 NLS_LENGTH_SEMANTICS 值。如果值不同（BYTE 与 CHAR），则可能是问题的原因。您可以通过在 VARCHAR2 中显式使用 BYTE 或 CHAR 来重建数据库之一上的类型和队列，即

create or replace type LOCAL_OBJ_MSG as object(
    test varchar2(4000 CHAR))
/

或更改其中一个数据库上的NLS_LENGTH_SEMANTICS以确保它们匹配。更改NLS_LENGTH_SEMANTICS后重新编译数据库很重要

于 2016-08-29T12:49:33.497 回答
0

我发布这个有点晚了，但我们有一个解决方法。在具有不同语义的两个数据库之间的传播中，您不能使用对象类型作为有效负载，但是您可以指定将该对象类型传播到可以传播的 xmltype 的转换。为此，您将需要临时或不需要远程数据库中的 queueu，它将接收该 xmltype。然后您可以有一个调度程序作业，该作业将定期从该队列中取出消息并将其进一步排入您的远程目标队列。在出队时，您可以指定另一个转换，它将您的 xmltype 有效负载转换为所需的对象类型。以下是示例：

假设我们有对象类型：

create or replace type OBJ_MSG as object(
    id number,
    value varchar2(4000)
)

我们将在本地和远程数据库中使用它。下一步是创建一个带有 xmltype 有效负载的远程队列。接下来我们需要转换函数，它将我们的 OBJ_MSG 对象类型作为参数并返回 xmltype。

function trans_to_xml_type(
  p_payload in OBJ_MSG)
  return xmltype is

  cursor c_trans_payload(
    cp_id number,
    cp_value varchar2) is
  select
    xmlelement(
      "envelope",
      xmlelement(
        "id", cp_client),
      xmlelement(
        "value", cp_value))
  from
    dual;

  l_return xmltype;
begin
  open c_trans_payload(
    p_payload.id,
    p_payload.value);
  fetch c_trans_payload
    into l_return;
  close c_trans_payload;

  return l_return;

  exception
    when others then
      raise_application_error(-20001, '.trans_to_xml_type .Failed to transform OBJ_MSG payload to sys.xmltype: ' || sqlerrm);
end;

接下来，我们创建我们的转换：

begin
  dbms_transform.create_transformation(
    'LOCAL',--SCHEMA
    'LOCAL_TRANS_TO_XML_TYPE',--Our transformation name
    'LOCAL',--SCHEMA
    'OBJ_MSG',--Our object name
    'SYS',
    'XMLTYPE',--Transformation result
    'PACKAGE.trans_to_xml_type(source.user_data)');--function name, which takes payload as a parameter
end;
/

接下来，我们使用我们的转换创建一个订阅者并将其添加到 queueu：

declare
  l_agent sys.aq$_agent := sys.aq$_agent(
    name => 'local_agent',
    address => 'REMOTE.REMOTE_TEMP_QUEUE@REMOTE_DB',
    protocol => 0);
begin
  dbms_aqadm.add_subscriber(
    queue_name => 'LOCAL.LOCAL_QUEUE',
    subscriber => l_agent,
    transformation => 'LOCAL_TRANS_TO_XML_TYPE',
    queue_to_queue => true);
end;
/

所以现在我们安排传播到我们的临时队列。传播消息的对象类型将转换为 xmltype 并排入我们的远程临时队列。在远程数据库中创建一个 dbms_scheduler 作业，该作业将使用指定的转换定期将消息出列并将其排入远程目标队列。

于 2018-02-07T10:58:47.837 回答