O gerenciamento eficiente de grandes volumes de dados \u00e9 crucial em sistemas de banco de dados. O Oracle Advanced Queue (AQ) oferece uma solu\u00e7\u00e3o robusta para processamento ass\u00edncrono e em paralelo de mensagens. Este artigo apresenta uma abordagem pr\u00e1tica para utilizar Oracle AQ em conjunto com threads em Java para processar registros de uma tabela particionada e subparticionada, onde o payload da mensagem \u00e9 um JSON.<\/p>\n
Vamos considerar uma tabela de pedidos ( Para gerenciar a fila, precisamos criar um tipo de mensagem JSON, uma tabela de filas e a fila propriamente dita.<\/p>\n Certifique-se de adicionar as depend\u00eancias do Spring JMS e do Oracle JDBC ao seu A classe Classe Repository<\/strong>:<\/p>\n A classe Classe Service<\/strong>:<\/p>\n Adicione a configura\u00e7\u00e3o do DataSource e do JMS no arquivo Vamos implementar a classe orders<\/code>) particionada por uma fun\u00e7\u00e3o MOD(256, id)<\/code> e subparticionada pelo dia do ano. A estrutura da tabela \u00e9 definida da seguinte maneira:<\/p>\n<\/span><\/code>CREATE TABLE orders (\r\n order_id NUMBER,\r\n customer_id NUMBER,\r\n order_date DATE,\r\n amount NUMBER,\r\n partition_id NUMBER GENERATED ALWAYS AS (MOD(order_id, 128)),\r\n partition_day NUMBER GENERATED ALWAYS AS (TO_NUMBER(TO_CHAR(order_date, 'DDD')))\r\n)\r\nPARTITION BY RANGE (partition_day)\r\nSUBPARTITION BY LIST (partition_id)\r\nSUBPARTITION TEMPLATE (\r\n SUBPARTITION p000 VALUES (0),\r\n SUBPARTITION p001 VALUES (1),\r\n -- Continue at\u00e9 p127\r\n SUBPARTITION p127 VALUES (127)\r\n)\r\n(\r\n PARTITION d001 VALUES LESS THAN (2),\r\n PARTITION d002 VALUES LESS THAN (3),\r\n -- Continue at\u00e9 d365\r\n PARTITION d365 VALUES LESS THAN (366)\r\n);\r\n<\/code><\/pre>\n<\/div>\nConfigurando o Oracle Advanced Queue<\/h4>\n
Cria\u00e7\u00e3o da Fila<\/h5>\n
-- Conectar-se ao esquema apropriado\r\nCONNECT your_schema\/your_password;\r\n\r\n-- Criar um tipo de objeto para a mensagem\r\nCREATE OR REPLACE TYPE message_type AS OBJECT (\r\n order_id NUMBER,\r\n customer_id NUMBER,\r\n order_date DATE,\r\n amount NUMBER\r\n);\r\n\r\n-- Criar uma tabela de filas\r\nBEGIN\r\n DBMS_AQADM.CREATE_QUEUE_TABLE(\r\n queue_table => 'order_queue_table',\r\n queue_payload_type => 'message_type'\r\n );\r\nEND;\r\n\/\r\n\r\n-- Criar a fila\r\nBEGIN\r\n DBMS_AQADM.CREATE_QUEUE(\r\n queue_name => 'order_queue',\r\n queue_table => 'order_queue_table'\r\n );\r\nEND;\r\n\/\r\n\r\n-- Iniciar a fila\r\nBEGIN\r\n DBMS_AQADM.START_QUEUE(\r\n queue_name => 'order_queue'\r\n );\r\nEND;\r\n\/\r\n<\/code><\/pre>\n<\/div>\nImplementa\u00e7\u00e3o em Java com Spring<\/h3>\n
Estrutura do Projeto<\/h4>\n
\n
Depend\u00eancias Maven<\/h5>\n
pom.xml<\/code>:<\/p>\n<dependency<\/span>>\r\n<\/span> <groupId<\/span>><\/span>org.springframework.boot<\/groupId<\/span>>\r\n<\/span> <artifactId<\/span>><\/span>spring-boot-starter-data-jpa<\/artifactId<\/span>>\r\n<\/span><\/dependency<\/span>>\r\n<\/span><dependency<\/span>>\r\n<\/span> <groupId<\/span>><\/span>org.springframework.boot<\/groupId<\/span>>\r\n<\/span> <artifactId<\/span>><\/span>spring-boot-starter-jms<\/artifactId<\/span>>\r\n<\/span><\/dependency<\/span>>\r\n<\/span><dependency<\/span>>\r\n<\/span> <groupId<\/span>><\/span>com.oracle.database.jdbc<\/groupId<\/span>>\r\n<\/span> <artifactId<\/span>><\/span>ojdbc8<\/artifactId<\/span>>\r\n<\/span> <version<\/span>><\/span>19.8.0.0<\/version<\/span>>\r\n<\/span><\/dependency<\/span>><\/span><\/code><\/pre>\n<\/div>\n1. Classe Repository<\/h4>\n
OrderRepository<\/code> gerencia a intera\u00e7\u00e3o com o banco de dados.<\/p>\nimport org.springframework.stereotype.Repository;\r\nimport javax.sql.DataSource;\r\nimport java.sql.*;\r\n\r\n@Repository\r\npublic class OrderRepository {\r\n private final DataSource dataSource;\r\n\r\n public OrderRepository(DataSource dataSource) {\r\n this.dataSource = dataSource;\r\n }\r\n\r\n public void insertOrder(int orderId, int customerId, Date orderDate, double amount) throws SQLException {\r\n String insertSQL = \"INSERT INTO orders (order_id, customer_id, order_date, amount) VALUES (?, ?, ?, ?)\";\r\n try (Connection conn = dataSource.getConnection();\r\n PreparedStatement stmt = conn.prepareStatement(insertSQL)) {\r\n stmt.setInt(1, orderId);\r\n stmt.setInt(2, customerId);\r\n stmt.setDate(3, orderDate);\r\n stmt.setDouble(4, amount);\r\n stmt.executeUpdate();\r\n }\r\n }\r\n\r\n public void enqueueMessage(int orderId, int customerId, Date orderDate, double amount) throws SQLException {\r\n String jsonMessage = String.format(\"{\\\"order_id\\\": %d, \\\"customer_id\\\": %d, \\\"order_date\\\": \\\"%s\\\", \\\"amount\\\": %f}\", orderId, customerId, orderDate, amount);\r\n String enqueueSQL = \"BEGIN \"\r\n + \"DBMS_AQ.ENQUEUE(:1, DBMS_AQ.ENQUEUE_OPTIONS_T(), DBMS_AQ.MESSAGE_PROPERTIES_T(), SYS.ANYDATA.ConvertVarchar2(:2), :3); \"\r\n + \"END;\";\r\n try (Connection conn = dataSource.getConnection();\r\n CallableStatement stmt = conn.prepareCall(enqueueSQL)) {\r\n stmt.setString(1, \"order_queue\");\r\n stmt.setString(2, jsonMessage);\r\n stmt.registerOutParameter(3, Types.RAW);\r\n stmt.execute();\r\n }\r\n }\r\n\r\n public String dequeueMessage() throws SQLException {\r\n String dequeueSQL = \"BEGIN \"\r\n + \"DBMS_AQ.DEQUEUE(:1, DBMS_AQ.DEQUEUE_OPTIONS_T(), DBMS_AQ.MESSAGE_PROPERTIES_T(), :2, :3); \"\r\n + \"END;\";\r\n try (Connection conn = dataSource.getConnection();\r\n CallableStatement stmt = conn.prepareCall(dequeueSQL)) {\r\n stmt.setString(1, \"order_queue\");\r\n stmt.registerOutParameter(2, Types.STRUCT, \"SYS.ANYDATA\");\r\n stmt.registerOutParameter(3, Types.RAW);\r\n stmt.execute();\r\n return stmt.getString(2);\r\n }\r\n }\r\n}\r\n<\/code><\/pre>\n<\/div>\n\n
orders<\/code>.<\/li>\n2. Classe Service<\/h4>\n
OrderService<\/code> utiliza o OrderRepository<\/code> para inserir pedidos e processar mensagens.<\/p>\nimport org.springframework.stereotype.Service;\r\nimport org.springframework.transaction.annotation.Transactional;\r\nimport java.sql.Date;\r\nimport java.sql.SQLException;\r\n\r\n@Service\r\npublic class OrderService {\r\n private final OrderRepository repository;\r\n\r\n public OrderService(OrderRepository repository) {\r\n this.repository = repository;\r\n }\r\n\r\n @Transactional\r\n public void insertOrder(int orderId, int customerId, Date orderDate, double amount) {\r\n try {\r\n repository.insertOrder(orderId, customerId, orderDate, amount);\r\n repository.enqueueMessage(orderId, customerId, orderDate, amount);\r\n } catch (SQLException e) {\r\n e.printStackTrace();\r\n throw new RuntimeException(e);\r\n }\r\n }\r\n\r\n public void processMessage(String message) {\r\n \/\/ Converter a mensagem JSON para um objeto Java\r\n JSONObject jsonObject = new JSONObject(message);\r\n int orderId = jsonObject.getInt(\"order_id\");\r\n int customerId = jsonObject.getInt(\"customer_id\");\r\n String orderDate = jsonObject.getString(\"order_date\");\r\n double amount = jsonObject.getDouble(\"amount\");\r\n\r\n \/\/ Processar a mensagem\r\n System.out.println(\"Processing order ID: \" + orderId + \", Customer ID: \" + customerId);\r\n \/\/ Adicionar a l\u00f3gica de processamento da mensagem aqui\r\n }\r\n\r\n public void processMessages() {\r\n try {\r\n while (true) {\r\n String message = repository.dequeueMessage();\r\n if (message != null) {\r\n System.out.println(\"Processing message: \" + message);\r\n \/\/ Adicionar l\u00f3gica de processamento de mensagens aqui\r\n } else {\r\n System.out.println(\"No messages to process.\");\r\n }\r\n }\r\n } catch (SQLException e) {\r\n e.printStackTrace();\r\n }\r\n }\r\n}\r\n<\/code><\/pre>\n<\/div>\n\n
OrderRepository<\/code> para inserir um pedido e enfileirar uma mensagem na fila.<\/li>\nOrderRepository<\/code>.<\/li>\n<\/ul>\n3. Configura\u00e7\u00e3o do JMS com Spring<\/h4>\n
Configura\u00e7\u00e3o do Application.yml<\/h5>\n
application.yml<\/code>:<\/p>\nspring:\r\n datasource:\r\n url: jdbc:oracle:thin:@\/\/hostname:port\/service_name\r\n username: your_schema\r\n password: your_password\r\n jms:\r\n listener:\r\n concurrency: 5-10\r\npartition:\r\n<\/span> id:<\/span> 1<\/span> # Defina o ID da parti\u00e7\u00e3o para este consumidor<\/span>\r\n<\/code><\/pre>\n<\/div>\nClasse de Configura\u00e7\u00e3o do JMS<\/h5>\n
import org.springframework.context.annotation.Bean;\r\nimport org.springframework.context.annotation.Configuration;\r\nimport org.springframework.jms.annotation.EnableJms;\r\nimport org.springframework.jms.config.DefaultJmsListenerContainerFactory;\r\nimport org.springframework.jms.core.JmsTemplate;\r\nimport org.apache.activemq.ActiveMQConnectionFactory;\r\n\r\n@Configuration\r\n@EnableJms\r\npublic class JmsConfig {\r\n @Bean\r\n public ActiveMQConnectionFactory connectionFactory() {\r\n return new ActiveMQConnectionFactory(\"tcp:\/\/localhost:61616\");\r\n }\r\n\r\n @Bean\r\n public DefaultJmsListenerContainerFactory jmsListenerContainerFactory() {\r\n DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();\r\n factory.setConnectionFactory(connectionFactory());\r\n factory.setConcurrency(\"5-10\");\r\n return factory;\r\n }\r\n\r\n @Bean\r\n public JmsTemplate jmsTemplate() {\r\n return new JmsTemplate(connectionFactory());\r\n }\r\n}\r\n<\/code><\/pre>\n<\/div>\nExplica\u00e7\u00e3o dos Componentes<\/h3>\n
\n
ActiveMQConnectionFactory<\/code><\/strong>:\n\n
61616<\/code>.<\/li>\nDefaultJmsListenerContainerFactory<\/code><\/strong>:\n\n
\"5-10\"<\/code>).<\/li>\nCLIENT_ACKNOWLEDGE<\/code>, onde o consumidor confirma explicitamente a mensagem ap\u00f3s o processamento.<\/li>\n<\/ul>\n<\/li>\nJmsTemplate<\/code><\/strong>:\n\n
Classe JMSConsumer<\/h5>\n
JMSConsumer<\/code> usando o Spring JMS. O componente JMSConsumer<\/code> usa a anota\u00e7\u00e3o @JmsListener<\/code> para receber e processar mensagens da fila. Com a configura\u00e7\u00e3o de concorr\u00eancia definida, v\u00e1rias threads podem processar mensagens em paralelo.<\/p>\nimport org.springframework.beans.factory.annotation.Autowired;\r\nimport org.springframework.jms.annotation.JmsListener;\r\nimport org.springframework.stereotype.Component;\r\n\r\n@Component\r\npublic class JMSConsumer {\r\n\r\n private final OrderService orderService;\r\n private<\/span> final<\/span> int<\/span> partitionId;\r\n\r\n @Autowired\r\n public JMSConsumer(OrderService orderService,