Como enviar mensagens multicast com SQL Server ServiceBroker

Começando com o SQL Server 11, o verbo SEND apresenta uma nova sintaxe e aceita múltiplos manipuladores de diálogos para enviar:

SEND
   ON CONVERSATION [(]conversation_handle [,.. @conversation_handle_n][)]
   [ MESSAGE TYPE message_type_name ]
   [ ( message_body_expression ) ]
[ ; ]

Com essa melhora de sintaxe, você pode enviar uma mensagem para vários destinos. Isso não é diferente de enviar a mesma mensagem várias vezes. Do ponto de vista da aplicação, a emissão de um único SEND em 10 manipuladores de diálogo é exatamente o mesmo que emitir 10 instruções SEND em um manipulador de diálogo por vez. A melhoria está no sys.transmission_queue: emitir o SEND várias vezes criaria diversas cópias do corpo da mensagem a ser enviada. Em contrapartida, um único SEND em múltiplos manipuladores armazenará o corpo da mensagem somente uma vez. Nós podemos ver isso se olharmos para a definição de sys.tranmission_queue no SQL Server 11:

p_helptext 'sys.transmission_queue'

CREATE VIEW sys.transmission_queue AS
 SELECT conversation_handle = S.handle,
  to_service_name = Q.tosvc,
  to_broker_instance = Q.tobrkrinst,
  from_service_name = Q.fromsvc,
  service_contract_name = Q.svccontr,
  enqueue_time = Q.enqtime,
  message_sequence_number = Q.msgseqnum,
  message_type_name = Q.msgtype,
  is_conversation_error = sysconv(bit, Q.status & 2),
  is_end_of_dialog = sysconv(bit, Q.status & 4),
  message_body = ISNULL(Q.msgbody, B.msgbody),
  transmission_status = GET_TRANSMISSION_STATUS (S.handle),
  priority = R.priority
 FROM sys.sysxmitqueue Q
 LEFT JOIN sys.sysxmitbody B WITH (NOLOCK) ON Q.msgref = B.msgref
 INNER JOIN sys.sysdesend S WITH (NOLOCK)
             ON Q.dlgid = S.diagid AND Q.finitiator = S.initiator
 INNER JOIN sys.sysdercv R WITH (NOLOCK)
             ON Q.dlgid = R.diagid AND Q.finitiator = R.initiator
 WHERE is_member('db_owner') = 1

Compare isso com a mesma definição de view no SQL Server 2008 R2:

CREATE VIEW sys.transmission_queue AS
 SELECT conversation_handle = S.handle,
  to_service_name = Q.tosvc,
  to_broker_instance = Q.tobrkrinst,
  from_service_name = Q.fromsvc,
  service_contract_name = Q.svccontr,
  enqueue_time = Q.enqtime,
  message_sequence_number = Q.msgseqnum,
  message_type_name = Q.msgtype,
  is_conversation_error = sysconv(bit, Q.status & 2),
  is_end_of_dialog = sysconv(bit, Q.status & 4),
  message_body = Q.msgbody,
  transmission_status = GET_TRANSMISSION_STATUS (S.handle),
  priority = R.priority
 FROM sys.sysxmitqueue Q
 INNER JOIN sys.sysdesend S WITH (NOLOCK)
               ON Q.dlgid = S.diagid AND Q.finitiator = S.initiator
 INNER JOIN sys.sysdercv R WITH (NOLOCK)
               ON Q.dlgid = R.diagid AND Q.finitiator = R.initiator
 WHERE is_member('db_owner') = 1

Você pode ver como no SQL Server 11 o corpo da mensagem foi separado em uma nova tabela de sistema (sys.sysxmitbody). O SEND multicast irá criar múltiplas entradas em sys.sysxmitqueue (uma para cada diálogo no qual a mensagem foi enviada por multicast), mas apenas uma entrada em sys.sysxmitbody. Tal esquema de armazenamento normalizado economiza o espaço consumido e, mais importante, a quantidade de log gerado durante o SEND.

O padrão de diálogo reverso em Publish-Subscribe

O padrão de diálogo típico em sistemas pub-sub é para que os assinantes iniciem o diálogo e enviem uma mensagem inicial ‘subscribe’, de forma que o conteúdo da assinatura está sendo entregue a partir do alvo (publicador/publisher) para o iniciador (assinante/subscriber). Eu chamo isso de padrão de diálogo reverso porque as mensagens fluem do publicador para o assinante. Vamos mostrar com um exemplo. Nós vamos criar um publisher service que envia por broadcast algum conteúdo importante, no qual serviços podem se inscrever para recebê-lo. Para apimentar, nós vamos usar um sistema de tag para nos inscrever em conteúdos opcionais: todo conteúdo é distribuído com uma lista de tags associadas, todos os assinantes especificam a tag na qual eles estão interessados. A correspondência das tags é feita usando a sintaxe LIKE, de forma que os assinantes possam especificar ‘%’ para se inscreverem em todo o conteúdo.

Publisher Service

create message type subscription_request validation = none;
create message type subscription_content validation = well_formed_xml;

create contract distribution
 (subscription_request sent by initiator,
 subscription_content sent by target);

create queue publisher;
create service publisher on queue publisher (distribution);
go

create table subscriptions (
 subscription_id int not null identity(1,1),
 tag nvarchar(50) not null,
 conversation_handle uniqueidentifier not null,
 constraint pk_subscriptions primary key (subscription_id),
 constraint unq_conversation_handle unique (conversation_handle, tag));
go

create procedure usp_publisher_handler
as
begin
 declare @mt sysname, @dh uniqueidentifier, @mb varbinary(max);
 begin try
  begin transaction;
  receive top(1)
   @mt = message_type_name,
   @dh = conversation_handle,
   @mb = message_body
   from publisher;
  if (@mt = N'subscription_request')
  begin
   insert into subscriptions (conversation_handle, tag)
                                   values (@dh, cast(@mb as nvarchar(50)));
  end
  else if(@mt = N'http://schemas.microsoft.com/SQL/ServiceBroker/Error'
   or @mt = N'http://schemas.microsoft.com/SQL/ServiceBroker/EndDialog')
  begin
   delete from subscriptions
    where conversation_handle  = @dh;
   end conversation @dh;
  end
  commit
 end try
 begin catch
  declare @xact_state int = xact_state();
  if @xact_state <> 0
  begin
   rollback;
  end
 end catch
end
go

alter queue publisher with activation (
 status = on,
 max_queue_readers = 1,
 procedure_name = usp_publisher_handler,
 execute as owner);
go

O publisher service é simples: ele usa uma tabela de assinaturas para manter o controle dos assinantes. O procedimento ativado associado com o publisher service processa as mensagens subscription_request e adiciona o diálogo de requisição à tabela de assinaturas. O corpo da mensagem de requisição é a tag em que o assinante está interessado.

O procedimento de publicação de conteúdo

create type publish_tags_type as table (
 tag nvarchar(50) not null primary key);
go

create procedure usp_publish_content
 @content xml,
 @tags publish_tags_type readonly
as
begin
 declare @sql nvarchar(max) = N'send on conversation (';
 declare @cnt int = 0;
 declare @dh uniqueidentifier;
 declare @comma nvarchar(2) = N'';

 declare crs cursor static read_only forward_only for
  select distinct conversation_handle
  from subscriptions s
  join @tags t on t.tag like s.tag;

 open crs;
 fetch next from crs into @dh;
 while 0 = @@fetch_status
 begin

  set @sql += @comma  + N'''' + cast(@dh as nvarchar(36)) + N'''';
  set @comma = N', ';
  set @cnt += 1;
  fetch next from crs into @dh;
 end
 close crs;
 deallocate crs;

 if @cnt > 0
 begin
  set @sql+= N') message type subscription_content (@content)';
  exec sp_executesql @sql, N'@content xml', @content;
 end
end
go

O procedimento de publicação de conteúdo recebe um conteúdo a ser distribuído e a lista de tags a partir das quais o conteúdo é distribuído, e envia o conteúdo a todos os assinantes interessados. Um único multicast SEND é usado para atingir todos os assinantes. Dynamic SQL é usado para construir a instrução multicast SEND.

Adicionando assinantes

declare @i int = 0;
declare @sql nvarchar(max);
while @i < 10
begin
 set @sql = N'create queue subscriber_' + cast(@i as nvarchar(20)) + N';
  create service subscriber_' + cast(@i as nvarchar(20)) + N'
                            on queue subscriber_'+cast(@i as nvarchar(20)) + N';';
 exec sp_executesql @sql;
 set @sql = N'declare @dh uniqueidentifier;
  begin dialog @dh
   from service subscriber_' + cast(@i as nvarchar(20)) + N'
   to service N''publisher''
   on contract distribution
   with encryption = off;
  send on conversation @dh message type subscription_request
                    (''' +case @i%5 when 0 then N'%' else nchar(@i + 65) end + ''');';
 exec sp_executesql @sql;
 set @i += 1;
end
go

Esse fragmento adiciona 10 assinantes interessados nas tags ‘B’, ‘C’, ‘D’ etc. O primeiro e o sexto assinantes estão interessados em tudo (‘%’). Nós podemos ver que os assinantes foram adicionados à tabela de assinaturas pelo procedimento de publicação ativado:

select * from subscriptions

subscription_id tag                               conversation_handle
--------------- ----------------- -------------------------------------
1               %                          AFC62EF2-35B3-E011-8EED-001C25160E57
2               B                          B3C62EF2-35B3-E011-8EED-001C25160E57
3               C                          B7C62EF2-35B3-E011-8EED-001C25160E57
4               D                          BBC62EF2-35B3-E011-8EED-001C25160E57
5               E                          BFC62EF2-35B3-E011-8EED-001C25160E57
6               %                          C3C62EF2-35B3-E011-8EED-001C25160E57
7               G                          C7C62EF2-35B3-E011-8EED-001C25160E57
8               H                          CBC62EF2-35B3-E011-8EED-001C25160E57
9               I                          CFC62EF2-35B3-E011-8EED-001C25160E57
10              J                          D3C62EF2-35B3-E011-8EED-001C25160E57

Uma mensagem multicast de teste

declare @tags publish_tags_type;
insert into @tags (tag) values ('A'), ('B'), ('C');
exec usp_publish_content N'', @tags;
go

Com essa única chamada, nós notificamos todos os assinantes interessados, com um único multicast SEND. Nós podemos verificar quais dos assinantes receberam o conteúdo:

declare @i int = 0;
declare @sql nvarchar(max) = N'', @union nvarchar(20) = N'';
while @i < 10
begin
 set @sql += @union + N'select
              ''subscriber_' + cast(@i as nvarchar(20)) + N''' as subscriber,
               count(*) as count
               from subscriber_' + cast(@i as nvarchar(20));
 set @union = ' union all ';
 set @i += 1;
end
exec sp_executesql @sql;
go

subscriber   count
------------ -----------
subscriber_0 1
subscriber_1 1
subscriber_2 1
subscriber_3 0
subscriber_4 0
subscriber_5 1
subscriber_6 0
subscriber_7 0
subscriber_8 0
subscriber_9 0

(10 row(s) affected)

Nós podemos ver que o subscriber_1 e o subscriber_2 receberam uma mensagem cada, já que as tags em que eles estão interessados são ‘B’ e ‘C’, e ambas correspondem ao conjunto de tags estipulado pelo publicador. Os assinantes 1 e 5 receberam uma mensagem cada, porque eles estão interessados em qualquer tag.

Esse padrão publish-subscribe não é novo e aplicações similares poderiam ser construídas com SQL Server Service Broker em SQL Server 2005, 2008 e 2008R2. Mas, com SQL Server 11, a distribuição é mais eficiente e pode escalar e funcionar melhor, já que os corpos das mensagens não são inseridos e removidos múltiplas vezes, uma para cada assinante, na fila de transmissão do publicador.

?

Texto original disponível em http://rusanu.com/2011/07/20/how-to-multicast-messages-with-sql-server-service-broker/