Fila em Banco: Escalando sua Aplicação com Banco Relacional

Um dos principais pilares para escalar uma aplicação web é o uso mensageria, não à toa temos diversos Brokers para enfileirar as mensagens a fim de processá-las num ritmo que nossa aplicação consiga atender.

Mas será mesmo que precisamos adotar um Broker? A verdade é que não; por mais controverso que pareça, podemos usar um banco de dados relacional como uma fila de mensagens, aproveitando sua natureza ACID e todo seu controle de locking e concorrência. Com isso, obtemos um alto grau de performance e principalmente escalabilidade sem a necessidade de abrir mão do poder de um banco de dados ou ter que adicionar um novo componente de software (como um JMS Broker) na arquitetura da solução.

Nessa talk, você vai aprender como implementar mensageria utilizando um RDBMS através da feature SKIP LOCKED e entender que escalabilidade não se resume a Kafka ou Brokers MQ.

(GRAVAÇÃO: https://www.youtube.com/watch?v=8DVFc7gXfIQ)

Transcript

1. FILA EM BANCO escalando sua aplicação com banco relacional
5. Hype-Driven Development https://blog.daftcode.pl/hype-driven-development-3469fc2e9b22
7. CACHE
8. CACHE PROCESSAMENTO ASSINCRONO
9. CACHE BALANCEAMENTO DE CARGA PROCESSAMENTO ASSINCRONO
10. CACHE BALANCEAMENTO DE CARGA PROCESSAMENTO ASSINCRONO
11. a idéia é simples…
12. servidor
13. db servidor
14. navegador db servidor
15. navegador db servidor requisição
16. navegador db servidor requisição resposta
17. e funciona muito bem…
18. db servidor requisição resposta
19. db servidor requisição resposta
20. db servidor requisição resposta
21. db servidor requisição resposta
22. calma!
23. navegador db servidor requisição resposta
24. navegador db servidor requisição resposta ???
25. navegador db servidor requisição resposta
26. navegador db servidor requisição resposta fila
27. navegador db servidor requisição resposta fila consumidor
28. navegador db servidor requisição resposta fila consumidor
29. navegador db servidor requisição resposta fila consumidor
30. mas qual broker?
35. mas qual o custo?
36. mas qual o custo? Segurança Backup Replicação Monitoramento Upgrade Recovery Fail-over Alta-disponibilidade Hardware
37. E agora?
38. Banco de dados
39. Sua app precisa dessa escala TODA?
40. Como escalar seu sistema usando FILA EM BANCO
41. Rafael Ponte @rponte
43. Fortaleza - Terra do Sol
45. FILA EM BANCO escalando sua aplicação com banco relacional
46. nosso problema
51. create package body PDF_Printer_Pkg as procedure print_pdf(p_xml varchar2) as begin end; end;
52. create package body PDF_Printer_Pkg as procedure print_pdf(p_xml varchar2) as begin end; end;
53. create package body PDF_Printer_Pkg as procedure print_pdf(p_xml varchar2) as begin end; end;
54. create package body PDF_Printer_Pkg as procedure print_pdf(p_xml varchar2) as begin end; end;
55. create package body PDF_Printer_Pkg as procedure print_pdf(p_xml varchar2) as begin end; end;
56. create package body PDF_Printer_Pkg as procedure print_pdf(p_xml varchar2) as begin end; end;
57. create package body PDF_Printer_Pkg as procedure print_pdf(p_xml varchar2) as begin -- valida XML end; end;
58. create package body PDF_Printer_Pkg as procedure print_pdf(p_xml varchar2) as begin -- valida XML -- processa XML e carrega dados end; end;
59. create package body PDF_Printer_Pkg as procedure print_pdf(p_xml varchar2) as begin -- valida XML -- processa XML e carrega dados -- gera PDF end; end;
60. create package body PDF_Printer_Pkg as procedure print_pdf(p_xml varchar2) as begin -- valida XML -- processa XML e carrega dados -- gera PDF -- faz upload para disco remoto end; end;
61. create package body PDF_Printer_Pkg as procedure print_pdf(p_xml varchar2) as begin -- valida XML -- processa XML e carrega dados -- gera PDF -- faz upload para disco remoto -- envia por email end; end;
62. create package body PDF_Printer_Pkg as procedure print_pdf(p_xml varchar2) as begin -- valida XML -- processa XML e carrega dados -- gera PDF -- faz upload para disco remoto -- envia por email end; end;
63. db
64. db servidor
65. db servidor navegador
66. db servidor navegador requisição
67. db servidor navegador requisição
68. db servidor navegador requisição
69. db servidor navegador requisição
70. db servidor requisição
71. db servidor requisição
72. db servidor requisição
73. db servidor requisição servidor servidor
74. db servidor requisição servidor servidor
76. vamos rodar em BACKGROUND !!!
77. COMO?
78. a gente joga numa FILA !
79. 4 3 2 QUEUE (fila)
80. 4 3 2 5 QUEUE (fila)
81. 4 3 2 5 enqueue QUEUE (fila)
82. 4 3 2 5 1 enqueue QUEUE (fila)
83. 4 3 2 5 1 enqueue dequeue QUEUE (fila)
84. 4 3 2 5 1 enqueue dequeue QUEUE (fila) FIFO: First-in, First-out
85. QUEUE (no mundo enterprise)
86. 1002 1001 QUEUE (no mundo enterprise)
87. 1002 1001 (message queue) QUEUE (no mundo enterprise)
88. 1002 1001 3506 QUEUE (no mundo enterprise) (message queue)
89. 1002 1001 3506 QUEUE (no mundo enterprise) (message) (message queue)
90. 1002 1001 producer QUEUE (no mundo enterprise) (message queue)
91. 1002 1001 producer 1000 QUEUE (no mundo enterprise) (message queue)
92. 1002 1001 producer consumer QUEUE (no mundo enterprise) (message queue)
93. ONDE?
94. no BANCO, ué!
95. tabela
96. db servidor navegador
97. db servidor navegador fila
98. db servidor navegador fila x
99. db servidor navegador requisição fila
100. db servidor navegador requisição INSERT fila
101. db servidor navegador requisição INSERT fila resposta
102. db servidor navegador requisição INSERT fila
103. db servidor navegador requisição INSERT quem executa? fila
104. db servidor navegador requisição INSERT fila
105. db servidor navegador requisição INSERT fila
106. e a vantagem?
107. db servidor requisição INSERT
108. db servidor requisição INSERT
109. db servidor requisição INSERT
110. db servidor requisição INSERT x
111. INSERT
112. INSERT
113. POLLING
114. como implementa?
115. precisamos de uma tabela…
116. describe job_queue;
117. Name Null? Type ----------- -------- ------------------- ID NOT NULL NUMBER CONTENT NOT NULL VARCHAR2(4000 CHAR) STATUS NOT NULL VARCHAR2(20 CHAR) THREAD_NAME VARCHAR2(60 CHAR) describe job_queue;
118. select j.* from job_queue j ;
119. select j.* from job_queue j ;
120. select j.* from job_queue j ;
121. select j.* from job_queue j ; ['NEW', 'FINISHED', 'ERROR']
122. e de um consumer…
123. create package body PDF_Printer_Pkg as procedure polling() as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = 'NEW'; begin for job in jobs_cursor loop -- processa item da fila print_pdf(job.content); -- atualiza status update job_queue j set j.status = 'FINISHED' where j.id = job.id; commit; -- comita a transação end loop; end; procedure print_pdf(p_xml varchar2) as begin -- ... end; end;
124. create package body PDF_Printer_Pkg as procedure polling() as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = 'NEW'; begin for job in jobs_cursor loop -- processa item da fila print_pdf(job.content); -- atualiza status update job_queue j set j.status = 'FINISHED' where j.id = job.id; commit; -- comita a transação end loop; end; procedure print_pdf(p_xml varchar2) as begin -- ... end; end;
125. create package body PDF_Printer_Pkg as procedure polling() as begin end; procedure print_pdf(p_xml varchar2) as begin -- ... end; end;
126. create package body PDF_Printer_Pkg as procedure polling() as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = 'NEW'; begin for job in jobs_cursor loop -- processa item da fila print_pdf(job.content); -- atualiza status update job_queue j set j.status = 'FINISHED' where j.id = job.id; commit; -- comita a transação end loop; end; procedure print_pdf(p_xml varchar2) as begin -- ... end; end;
127. create package body PDF_Printer_Pkg as procedure polling() as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = 'NEW'; begin for job in jobs_cursor loop -- processa item da fila print_pdf(job.content); -- atualiza status update job_queue j set j.status = 'FINISHED' where j.id = job.id; commit; -- comita a transação end loop; end; procedure print_pdf(p_xml varchar2) as begin -- ... end; end;
128. create package body PDF_Printer_Pkg as procedure polling() as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = 'NEW'; begin for job in jobs_cursor loop -- processa item da fila print_pdf(job.content); -- atualiza status update job_queue j set j.status = 'FINISHED' where j.id = job.id; commit; -- comita a transação end loop; end; procedure print_pdf(p_xml varchar2) as begin -- ... end; end;
129. create package body PDF_Printer_Pkg as procedure polling() as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = 'NEW'; begin for job in jobs_cursor loop -- processa item da fila print_pdf(job.content); -- atualiza status update job_queue j set j.status = 'FINISHED' where j.id = job.id; commit; -- comita a transação end loop; end; procedure print_pdf(p_xml varchar2) as begin -- ... end; end;
130. create package body PDF_Printer_Pkg as procedure polling() as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = 'NEW'; begin for job in jobs_cursor loop -- processa item da fila print_pdf(job.content); -- atualiza status update job_queue j set j.status = 'FINISHED' where j.id = job.id; commit; -- comita a transação end loop; end; procedure print_pdf(p_xml varchar2) as begin -- ... end; end;
131. create package body PDF_Printer_Pkg as procedure polling() as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = 'NEW'; begin for job in jobs_cursor loop -- processa item da fila print_pdf(job.content); -- atualiza status update job_queue j set j.status = 'FINISHED' where j.id = job.id; commit; -- comita a transação end loop; end; procedure print_pdf(p_xml varchar2) as begin -- ... end; end;
132. e alguém para fazer o polling…
133. BEGIN DBMS_SCHEDULER.create_job ( job_name => 'PDF_Printer' ,job_type => 'PLSQL_BLOCK' ,job_action => 'BEGIN PDF_Printer_Pkg.polling(); END;' ,start_date => systimestamp ,repeat_interval => 'freq=minutely;' ,enabled => true ); END;
134. BEGIN DBMS_SCHEDULER.create_job ( job_name => 'PDF_Printer' ,job_type => 'PLSQL_BLOCK' ,job_action => 'BEGIN PDF_Printer_Pkg.polling(); END;' ,start_date => systimestamp ,repeat_interval => 'freq=minutely;' ,enabled => true ); END;
135. BEGIN DBMS_SCHEDULER.create_job ( job_name => 'PDF_Printer' ,job_type => 'PLSQL_BLOCK' ,job_action => 'BEGIN PDF_Printer_Pkg.polling(); END;' ,start_date => systimestamp ,repeat_interval => 'freq=minutely;' ,enabled => true ); END;
136. BEGIN DBMS_SCHEDULER.create_job ( job_name => 'PDF_Printer' ,job_type => 'PLSQL_BLOCK' ,job_action => 'BEGIN PDF_Printer_Pkg.polling(); END;' ,start_date => systimestamp ,repeat_interval => 'freq=minutely;' ,enabled => true ); END;
137. INSERT
138. select j.* from job_queue j ;
139. select j.* from job_queue j ;
140. select j.* from job_queue j ;
141. select j.* from job_queue j ; Uhuuull!!
142. Mas e se…
143. 9630 9631 QUEUE (fila) producer consumer 9629
144. 9630 9631 QUEUE (fila) producer consumer 9629 produz: 1000 itens/min
145. 9630 9631 QUEUE (fila) producer consumer 9629 produz: 1000 itens/min consume: 250 itens/min
146. e agora?
147. Fácil! levanta outro CONSUMER!
148. 9630 9631 QUEUE (fila) producer consumer 9629 produz: 1000 itens/min
149. 9630 9631 QUEUE (fila) producer consumer 9629 produz: 1000 itens/min consumer consume: 250 itens/min consume: 250 itens/min
150. DISTRIBUTED POLLING
152. 1000 / 250
153. 1000 4 / 250 =
154. BEGIN DBMS_SCHEDULER.create_job ( job_name => 'PDF_Printer_1' ,job_action => 'BEGIN PDF_Printer_Pkg.polling(''thread_1''); END;' -- ... ); DBMS_SCHEDULER.create_job ( job_name => 'PDF_Printer_2' ,job_action => 'BEGIN PDF_Printer_Pkg.polling(''thread_2''); END;' -- ... ); DBMS_SCHEDULER.create_job ( job_name => 'PDF_Printer_3' ,job_action => 'BEGIN PDF_Printer_Pkg.polling(''thread_3''); END;' -- ... ); DBMS_SCHEDULER.create_job ( job_name => 'PDF_Printer_4' ,job_action => 'BEGIN PDF_Printer_Pkg.polling(‘'thread_4''); END;' -- ... ); END;
155. BEGIN DBMS_SCHEDULER.create_job ( job_name => 'PDF_Printer_1' ,job_action => 'BEGIN PDF_Printer_Pkg.polling(''thread_1''); END;' -- ... ); DBMS_SCHEDULER.create_job ( job_name => 'PDF_Printer_2' ,job_action => 'BEGIN PDF_Printer_Pkg.polling(''thread_2''); END;' -- ... ); DBMS_SCHEDULER.create_job ( job_name => 'PDF_Printer_3' ,job_action => 'BEGIN PDF_Printer_Pkg.polling(''thread_3''); END;' -- ... ); DBMS_SCHEDULER.create_job ( job_name => 'PDF_Printer_4' ,job_action => 'BEGIN PDF_Printer_Pkg.polling(''thread_4''); END;' -- ... ); END;
156. Name Null? Type ----------- -------- ------------------- ID NOT NULL NUMBER CONTENT NOT NULL VARCHAR2(4000 CHAR) STATUS NOT NULL VARCHAR2(20 CHAR) THREAD_NAME VARCHAR2(60 CHAR) describe job_queue;
157. procedure polling(p_thread_name varchar2) as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = 'NEW'; begin for job in jobs_cursor loop -- processa item da fila print_pdf(job.content); -- atualiza status update job_queue j set j.status = 'FINISHED' ,j.thread_name = p_thread_name where j.id = job.id; commit; -- comita a transação end loop; end;
158. INSERT
159. select j.* from job_queue j ;
160. select j.* from job_queue j ;
161. select j.* from job_queue j ;
162. select j.* from job_queue j ;
163. YES!
164. Calma aí…
165. …mas está mais LENTO
166. …e tem clientes recebendo VÁRIOS emails! …mas está mais LENTO
167. Vamos entender o que aconteceu…
168. ID STATUS THREAD_NAME 1001 NEW 1002 NEW 1003 NEW 1004 NEW
169. ID STATUS THREAD_NAME 1001 NEW 1002 NEW 1003 NEW 1004 NEW thread_1
170. ID STATUS THREAD_NAME 1001 NEW 1002 NEW 1003 NEW 1004 NEW thread_1 thread_2
171. ID STATUS THREAD_NAME 1001 NEW 1002 NEW 1003 NEW 1004 NEW thread_1 thread_2 select j.* from job_queue j where j.status = 'NEW';
172. ID STATUS THREAD_NAME 1001 NEW 1002 NEW 1003 NEW 1004 NEW thread_1 thread_2 select j.* from job_queue j where j.status = 'NEW'; {id=1001}
173. ID STATUS THREAD_NAME 1001 NEW 1002 NEW 1003 NEW 1004 NEW thread_1 thread_2 select j.* from job_queue j where j.status = 'NEW'; {id=1001} select j.* from job_queue j where j.status = 'NEW';
174. ID STATUS THREAD_NAME 1001 NEW 1002 NEW 1003 NEW 1004 NEW thread_1 thread_2 select j.* from job_queue j where j.status = 'NEW'; {id=1001} select j.* from job_queue j where j.status = 'NEW'; {id=1001}
175. ID STATUS THREAD_NAME 1001 FINISHED thread_1 1002 NEW 1003 NEW 1004 NEW thread_1 thread_2 select j.* from job_queue j where j.status = 'NEW'; {id=1001} update job_queue j set j.status = 'FINISHED' ,j.thread_name = 'thread_1' where j.id = 1001; select j.* from job_queue j where j.status = 'NEW'; {id=1001}
176. ID STATUS THREAD_NAME 1001 FINISHED thread_1 1002 NEW 1003 NEW 1004 NEW thread_1 thread_2 select j.* from job_queue j where j.status = 'NEW'; {id=1001} update job_queue j set j.status = 'FINISHED' ,j.thread_name = 'thread_1' where j.id = 1001; commit; select j.* from job_queue j where j.status = 'NEW'; {id=1001}
177. ID STATUS THREAD_NAME 1001 FINISHED thread_1 1002 NEW 1003 NEW 1004 NEW thread_1 thread_2 select j.* from job_queue j where j.status = 'NEW'; {id=1001} select j.* from job_queue j where j.status = 'NEW'; {id=1001} update job_queue j set j.status = 'FINISHED' ,j.thread_name = 'thread_1' where j.id = 1001; commit; update job_queue j set j.status = 'FINISHED' ,j.thread_name = ‘thread_2' where j.id = 1001; commit; thread_2
178. ID STATUS THREAD_NAME 1001 FINISHED thread_1 1002 FINISHED thread_1 1003 NEW 1004 NEW thread_1 thread_2 select j.* from job_queue j where j.status = 'NEW'; {id=1002} select j.* from job_queue j where j.status = 'NEW'; {id=1002} update job_queue j set j.status = 'FINISHED' ,j.thread_name = 'thread_1' where j.id = 1002; commit; update job_queue j set j.status = 'FINISHED' ,j.thread_name = ‘thread_2' where j.id = 1002; commit; thread_2 thread_2
179. ID STATUS THREAD_NAME 1001 FINISHED thread_1 1002 FINISHED thread_1 1003 FINISHED thread_2 1004 NEW thread_1 thread_2 select j.* from job_queue j where j.status = 'NEW'; {id=1003} select j.* from job_queue j where j.status = 'NEW'; {id=1003} update job_queue j set j.status = 'FINISHED' ,j.thread_name = 'thread_1' where j.id = 1003; commit; update job_queue j set j.status = 'FINISHED' ,j.thread_name = ‘thread_2' where j.id = 1003; commit; thread_2 thread_1 thread_2
180. ID STATUS THREAD_NAME 1001 FINISHED thread_1 1002 FINISHED thread_1 1003 FINISHED thread_2 1004 FINISHED thread_1 thread_1 thread_2 select j.* from job_queue j where j.status = 'NEW'; {id=1004} select j.* from job_queue j where j.status = 'NEW'; {id=1004} update job_queue j set j.status = 'FINISHED' ,j.thread_name = 'thread_1' where j.id = 1004; commit; update job_queue j set j.status = 'FINISHED' ,j.thread_name = ‘thread_2' where j.id = 1004; commit; thread_2 thread_1 thread_2 thread_2
181. NO!
182. o desenho da solução deve ser pensado para multi-threaded * desenho e implementação *
183. e agora?
184. Podemos LOCKAR as linhas
185. procedure polling(p_thread_name varchar2) as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = 'NEW'; begin for job in jobs_cursor loop -- processa item da fila print_pdf(job.content); -- atualiza status update job_queue j set j.status = ‘FINISHED' ,j.thread_name = p_thread_name where j.id = job.id; commit; -- comita a transação end loop; end;
186. procedure polling(p_thread_name varchar2) as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = 'NEW'; begin for job in jobs_cursor loop -- processa item da fila print_pdf(job.content); -- atualiza status update job_queue j set j.status = 'FINISHED' ,j.thread_name = p_thread_name where j.id = job.id; commit; -- comita a transação end loop; end;
187. procedure polling(p_thread_name varchar2) as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = 'NEW'; begin for job in jobs_cursor loop -- processa item commit; end loop; end;
188. procedure polling(p_thread_name varchar2) as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = 'NEW'; begin for job in jobs_cursor loop -- processa item commit; end loop; end;
189. procedure polling(p_thread_name varchar2) as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = 'NEW' for update; begin for job in jobs_cursor loop -- processa item commit; end loop; end;
190. procedure polling(p_thread_name varchar2) as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = 'NEW' for update; begin for job in jobs_cursor loop -- processa item commit; end loop; end; locka TODOS os registros
191. procedure polling(p_thread_name varchar2) as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = 'NEW' and rownum = 1 for update; begin for job in jobs_cursor loop -- processa item commit; end loop; end;
192. procedure polling(p_thread_name varchar2) as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = 'NEW' and rownum = 1 for update; begin for job in jobs_cursor loop -- processa item commit; end loop; end; SELECT…FOR UPDATE não suporta FETCHING-ACROSS-COMMITS
193. procedure polling(p_thread_name varchar2) as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = ‘NEW’ and rownum = 1 for update; begin for job in jobs_cursor loop -- processa item commit; end loop; end; Vai sair muito cedo do loop
194. procedure polling(p_thread_name varchar2) as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = ‘NEW’ and rownum = 1 for update; begin for job in jobs_cursor loop -- processa item commit; end loop; end;
195. procedure polling(p_thread_name varchar2) as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = ‘NEW’ and rownum = 1 for update; begin for job in jobs_cursor loop -- processa item commit; end loop; end;
196. procedure polling(p_thread_name varchar2) as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = ‘NEW' and rownum = 1 for update; job jobs_cursor%rowtype; begin loop open jobs_cursor; fetch jobs_cursor into job; exit when jobs_cursor%notfound; -- processa item commit; close jobs_cursor; end loop; end;
197. procedure polling(p_thread_name varchar2) as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = ‘NEW' and rownum = 1 for update; job jobs_cursor%rowtype; begin loop open jobs_cursor; fetch jobs_cursor into job; exit when jobs_cursor%notfound; -- processa item commit; close jobs_cursor; end loop; end;
198. procedure polling(p_thread_name varchar2) as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = ‘NEW' and rownum = 1 for update; job jobs_cursor%rowtype; begin loop open jobs_cursor; fetch jobs_cursor into job; exit when jobs_cursor%notfound; -- processa item commit; close jobs_cursor; end loop; end;
199. procedure polling(p_thread_name varchar2) as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = ‘NEW' and rownum = 1 for update; job jobs_cursor%rowtype; begin loop open jobs_cursor; fetch jobs_cursor into job; exit when jobs_cursor%notfound; -- processa item commit; close jobs_cursor; end loop; end; lock ocorre no OPENING
200. procedure polling(p_thread_name varchar2) as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = ‘NEW' and rownum = 1 for update; job jobs_cursor%rowtype; begin loop open jobs_cursor; fetch jobs_cursor into job; exit when jobs_cursor%notfound; -- processa item commit; close jobs_cursor; end loop; end;
201. procedure polling(p_thread_name varchar2) as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = ‘NEW' and rownum = 1 for update; job jobs_cursor%rowtype; begin loop open jobs_cursor; fetch jobs_cursor into job; exit when jobs_cursor%notfound; -- processa item commit; close jobs_cursor; end loop; end;
202. procedure polling(p_thread_name varchar2) as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = ‘NEW' and rownum = 1 for update; job jobs_cursor%rowtype; begin loop open jobs_cursor; fetch jobs_cursor into job; exit when jobs_cursor%notfound; -- processa item commit; close jobs_cursor; end loop; end;
203. procedure polling(p_thread_name varchar2) as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = ‘NEW' and rownum = 1 for update; job jobs_cursor%rowtype; begin loop open jobs_cursor; fetch jobs_cursor into job; exit when jobs_cursor%notfound; -- processa item commit; close jobs_cursor; end loop; end;
204. procedure polling(p_thread_name varchar2) as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = 'NEW' and rownum = 1 for update; job jobs_cursor%rowtype; begin loop open jobs_cursor; fetch jobs_cursor into job; exit when jobs_cursor%notfound; -- processa item commit; close jobs_cursor; end loop; end;
205. INSERT
206. select j.* from job_queue j ;
207. select j.* from job_queue j ;
208. select j.* from job_queue j ;
209. select j.* from job_queue j ;
210. YES!
211. Calma aí…
212. …não há duplicação
213. …não há duplicação …mas ainda não está rápido o suficiente!
214. Vamos entender o que aconteceu…
215. select j.* from job_queue j where j.status = 'NEW' and rownum = 1 for update; select j.* from job_queue j where j.status = 'NEW' and rownum = 1 for update; thread_1 thread_2 ID STATUS THREAD_NAME 1001 NEW 1002 NEW 1003 NEW 1004 NEW
216. select j.* from job_queue j where j.status = 'NEW' and rownum = 1 for update; select j.* from job_queue j where j.status = 'NEW' and rownum = 1 for update; thread_1 thread_2 ID STATUS THREAD_NAME 1001 NEW 1002 NEW 1003 NEW 1004 NEW
217. select j.* from job_queue j where j.status = 'NEW' and rownum = 1 for update; select j.* from job_queue j where j.status = 'NEW' and rownum = 1 for update; thread_1 thread_2 ID STATUS THREAD_NAME 1001 NEW 1002 NEW 1003 NEW 1004 NEW
218. select j.* from job_queue j where j.status = 'NEW' and rownum = 1 for update; select j.* from job_queue j where j.status = 'NEW' and rownum = 1 for update; thread_1 thread_2 ID STATUS THREAD_NAME 1001 NEW 1002 NEW 1003 NEW 1004 NEW
219. update job_queue j set j.status = 'FINISHED' ,j.thread_name = 'thread_1' where j.id = 1001; select j.* from job_queue j where j.status = 'NEW' and rownum = 1 for update; thread_1 thread_2 ID STATUS THREAD_NAME 1001 FINISHED thread_1 1002 NEW 1003 NEW 1004 NEW
220. select j.* from job_queue j where j.status = 'NEW' and rownum = 1 for update; thread_1 thread_2 ID STATUS THREAD_NAME 1001 FINISHED thread_1 1002 NEW 1003 NEW 1004 NEW commit;
221. select j.* from job_queue j where j.status = 'NEW' and rownum = 1 for update; thread_1 thread_2 ID STATUS THREAD_NAME 1001 FINISHED thread_1 1002 NEW 1003 NEW 1004 NEW commit;
222. select j.* from job_queue j where j.status = 'NEW' and rownum = 1 for update; thread_1 thread_2 ID STATUS THREAD_NAME 1001 FINISHED thread_1 1002 NEW 1003 NEW 1004 NEW select j.* from job_queue j where j.status = 'NEW' and rownum = 1 for update;
223. select j.* from job_queue j where j.status = 'NEW' and rownum = 1 for update; thread_1 thread_2 ID STATUS THREAD_NAME 1001 FINISHED thread_1 1002 NEW 1003 NEW 1004 NEW select j.* from job_queue j where j.status = 'NEW' and rownum = 1 for update;
224. select j.* from job_queue j where j.status = 'NEW' and rownum = 1 for update; thread_1 thread_2 ID STATUS THREAD_NAME 1001 FINISHED thread_1 1002 NEW 1003 NEW 1004 NEW select j.* from job_queue j where j.status = 'NEW' and rownum = 1 for update;
225. update job_queue j set j.status = 'FINISHED' ,j.thread_name = ‘thread_2' where j.id = 1002; thread_1 thread_2 ID STATUS THREAD_NAME 1001 FINISHED thread_1 1002 FINISHED thread_2 1003 NEW 1004 NEW select j.* from job_queue j where j.status = 'NEW' and rownum = 1 for update;
226. commit; thread_1 thread_2 ID STATUS THREAD_NAME 1001 FINISHED thread_1 1002 FINISHED thread_2 1003 NEW 1004 NEW select j.* from job_queue j where j.status = 'NEW' and rownum = 1 for update;
227. commit; thread_1 thread_2 ID STATUS THREAD_NAME 1001 FINISHED thread_1 1002 FINISHED thread_2 1003 NEW 1004 NEW select j.* from job_queue j where j.status = 'NEW' and rownum = 1 for update;
228. select j.* from job_queue j where j.status = 'NEW' and rownum = 1 for update; thread_1 thread_2 ID STATUS THREAD_NAME 1001 FINISHED thread_1 1002 FINISHED thread_2 1003 NEW 1004 NEW select j.* from job_queue j where j.status = 'NEW' and rownum = 1 for update;
229. select j.* from job_queue j where j.status = 'NEW' and rownum = 1 for update; thread_1 thread_2 ID STATUS THREAD_NAME 1001 FINISHED thread_1 1002 FINISHED thread_2 1003 FINISHED thread_2 1004 NEW select j.* from job_queue j where j.status = 'NEW' and rownum = 1 for update;
230. select j.* from job_queue j where j.status = 'NEW' and rownum = 1 for update; thread_1 thread_2 ID STATUS THREAD_NAME 1001 FINISHED thread_1 1002 FINISHED thread_2 1003 FINISHED thread_2 1004 FINISHED thread_1 select j.* from job_queue j where j.status = 'NEW' and rownum = 1 for update;
231. select j.* from job_queue j where j.status = 'NEW' and rownum = 1 for update; thread_1 thread_2 ID STATUS THREAD_NAME 1001 FINISHED thread_1 1002 FINISHED thread_2 1003 FINISHED thread_2 1004 FINISHED thread_1 select j.* from job_queue j where j.status = 'NEW' and rownum = 1 for update;
232. humm…
233. e agora?
234. Pensando…
235. Que tal tentarmos o SKIP LOCKED?
236. procedure polling(p_thread_name varchar2) as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = 'NEW' and rownum = 1 for update; job jobs_cursor%rowtype; begin loop open jobs_cursor; fetch jobs_cursor into job; exit when jobs_cursor%notfound; -- processa item commit; close jobs_cursor; end loop; end;
237. procedure polling(p_thread_name varchar2) as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = 'NEW' and rownum = 1 for update skip locked; job jobs_cursor%rowtype; begin loop open jobs_cursor; fetch jobs_cursor into job; exit when jobs_cursor%notfound; -- processa item commit; close jobs_cursor; end loop; end;
238. procedure polling(p_thread_name varchar2) as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = 'NEW' and rownum = 1 for update skip locked; job jobs_cursor%rowtype; begin loop open jobs_cursor; fetch jobs_cursor into job; exit when jobs_cursor%notfound; -- processa item commit; close jobs_cursor; end loop; end; lock ocorre no FETCHING
239. procedure polling(p_thread_name varchar2) as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = 'NEW' and rownum = 1 for update skip locked; job jobs_cursor%rowtype; begin loop open jobs_cursor; fetch jobs_cursor into job; exit when jobs_cursor%notfound; -- processa item commit; close jobs_cursor; end loop; end; lock ocorre no FETCHING
240. procedure polling(p_thread_name varchar2) as cursor jobs_cursor is select jq.* from job_queue jq where jq.status = 'NEW' for update skip locked; job jobs_cursor%rowtype; begin loop open jobs_cursor; fetch jobs_cursor into job; exit when jobs_cursor%notfound; -- processa item commit; close jobs_cursor; end loop; end;
241. Qual a diferença?
242. select j.* from job_queue j where j.status = 'NEW' for update SKIP LOCKED; select j.* from job_queue j where j.status = 'NEW' for update SKIP LOCKED; thread_1 thread_2 ID STATUS THREAD_NAME 1001 NEW 1002 NEW 1003 NEW 1004 NEW
243. select j.* from job_queue j where j.status = 'NEW' for update SKIP LOCKED; select j.* from job_queue j where j.status = 'NEW' for update SKIP LOCKED; thread_1 thread_2 ID STATUS THREAD_NAME 1001 NEW 1002 NEW 1003 NEW 1004 NEW
244. select j.* from job_queue j where j.status = 'NEW' for update SKIP LOCKED; select j.* from job_queue j where j.status = 'NEW' for update SKIP LOCKED; thread_1 thread_2 ID STATUS THREAD_NAME 1001 NEW 1002 NEW 1003 NEW 1004 NEW
245. select j.* from job_queue j where j.status = 'NEW' for update SKIP LOCKED; select j.* from job_queue j where j.status = 'NEW' for update SKIP LOCKED; thread_1 thread_2 ID STATUS THREAD_NAME 1001 NEW 1002 NEW 1003 NEW 1004 NEW
246. select j.* from job_queue j where j.status = 'NEW' for update SKIP LOCKED; select j.* from job_queue j where j.status = 'NEW' for update SKIP LOCKED; thread_1 thread_2 ID STATUS THREAD_NAME 1001 NEW 1002 NEW 1003 NEW 1004 NEW
247. select j.* from job_queue j where j.status = 'NEW' for update SKIP LOCKED; select j.* from job_queue j where j.status = 'NEW' for update SKIP LOCKED; thread_1 thread_2 ID STATUS THREAD_NAME 1001 NEW 1002 NEW 1003 NEW 1004 NEW
248. select j.* from job_queue j where j.status = 'NEW' for update SKIP LOCKED; select j.* from job_queue j where j.status = 'NEW' for update SKIP LOCKED; thread_1 thread_2 ID STATUS THREAD_NAME 1001 NEW 1002 NEW 1003 NEW 1004 NEW
249. update job_queue j set j.status = 'FINISHED' ,j.thread_name = ‘thread_1' where j.id = 1001; update job_queue j set j.status = 'FINISHED' ,j.thread_name = 'thread_2' where j.id = 1002; thread_1 thread_2 ID STATUS THREAD_NAME 1001 FINISHED thread_1 1002 FINISHED thread_2 1003 NEW 1004 NEW
250. commit; update job_queue j set j.status = 'FINISHED' ,j.thread_name = 'thread_2' where j.id = 1002; thread_1 thread_2 ID STATUS THREAD_NAME 1001 FINISHED thread_1 1002 FINISHED thread_2 1003 NEW 1004 NEW
251. select j.* from job_queue j where j.status = 'NEW' for update SKIP LOCKED; commit; thread_1 thread_2 ID STATUS THREAD_NAME 1001 FINISHED thread_1 1002 FINISHED thread_2 1003 NEW 1004 NEW
252. select j.* from job_queue j where j.status = 'NEW' for update SKIP LOCKED; select j.* from job_queue j where j.status = 'NEW' for update SKIP LOCKED; thread_1 thread_2 ID STATUS THREAD_NAME 1001 FINISHED thread_1 1002 FINISHED thread_2 1003 NEW 1004 NEW
253. select j.* from job_queue j where j.status = 'NEW' for update SKIP LOCKED; select j.* from job_queue j where j.status = 'NEW' for update SKIP LOCKED; thread_1 thread_2 ID STATUS THREAD_NAME 1001 FINISHED thread_1 1002 FINISHED thread_2 1003 NEW 1004 NEW
254. select j.* from job_queue j where j.status = 'NEW' for update SKIP LOCKED; select j.* from job_queue j where j.status = 'NEW' for update SKIP LOCKED; thread_1 thread_2 ID STATUS THREAD_NAME 1001 FINISHED thread_1 1002 FINISHED thread_2 1003 NEW 1004 NEW
255. select j.* from job_queue j where j.status = 'NEW' for update SKIP LOCKED; select j.* from job_queue j where j.status = 'NEW' for update SKIP LOCKED; thread_1 thread_2 ID STATUS THREAD_NAME 1001 FINISHED thread_1 1002 FINISHED thread_2 1003 NEW 1004 NEW
256. Ou seja…
257. select j.* from job_queue j where j.status = 'NEW' for update SKIP LOCKED; select j.* from job_queue j where j.status = 'NEW' for update SKIP LOCKED; thread_1 thread_2 ID STATUS THREAD_NAME 1001 NEW 1002 NEW 1003 NEW 1004 NEW select j.* from job_queue j where j.status = 'NEW' for update SKIP LOCKED; select j.* from job_queue j where j.status = 'NEW' for update SKIP LOCKED; thread_3 thread_4
258. select j.* from job_queue j where j.status = 'NEW' for update SKIP LOCKED; select j.* from job_queue j where j.status = 'NEW' for update SKIP LOCKED; thread_1 thread_2 ID STATUS THREAD_NAME 1001 NEW 1002 NEW 1003 NEW 1004 NEW select j.* from job_queue j where j.status = 'NEW' for update SKIP LOCKED; select j.* from job_queue j where j.status = 'NEW' for update SKIP LOCKED; thread_3 thread_4
259. INSERT
260. INSERT
261. INSERT
262. YES!
263. Concluindo
264. single-threaded funciona bem…
265. single-threaded funciona bem… https://www.pgcon.org/2016/schedule/track/Applications/929.en.html
266. multi-threaded com FOR UPDATE…
267. multi-threaded com FOR UPDATE… https://www.pgcon.org/2016/schedule/track/Applications/929.en.html
268. multi-threaded com SKIP LOCKED…
269. multi-threaded com SKIP LOCKED… https://www.pgcon.org/2016/schedule/track/Applications/929.en.html
270. ESCALABILIDADE
271. (desde 10g) (desde 8.0) (desde 2005) (desde 9.5)
272. ORACLE AQ (Advanced Queue)
274. OBRIGADO!
275. @rponte Rafael Ponte