|
1
21 package net.sf.hajdbc.local;
22
23 import java.sql.Connection  ;
24 import java.sql.DatabaseMetaData ;
25 import java.sql.Driver ;
26 import java.sql.ResultSet ;
27 import java.sql.Statement ;
28 import java.util.ArrayList ;
29 import java.util.Collection ;
30 import java.util.HashMap ;
31 import java.util.Iterator ;
32 import java.util.LinkedList ;
33 import java.util.List ;
34 import java.util.Map ;
35 import java.util.Set ;
36 import java.util.TreeSet ;
37 import java.util.concurrent.ConcurrentHashMap ;
38 import java.util.concurrent.ExecutorService ;
39 import java.util.concurrent.SynchronousQueue ;
40 import java.util.concurrent.ThreadPoolExecutor ;
41 import java.util.concurrent.TimeUnit ;
42 import java.util.concurrent.locks.Lock ;
43 import java.util.concurrent.locks.ReadWriteLock ;
44 import java.util.concurrent.locks.ReentrantReadWriteLock ;
45 import java.util.prefs.BackingStoreException ;
46 import java.util.prefs.Preferences ;
47
48 import javax.management.JMException ;
49 import javax.management.MBeanServer ;
50 import javax.management.ObjectName ;
51 import javax.management.StandardMBean ;
52 import javax.sql.DataSource ;
53
54 import net.sf.hajdbc.Balancer;
55 import net.sf.hajdbc.Database;
56 import net.sf.hajdbc.DatabaseCluster;
57 import net.sf.hajdbc.DatabaseClusterFactory;
58 import net.sf.hajdbc.DatabaseClusterMBean;
59 import net.sf.hajdbc.Dialect;
60 import net.sf.hajdbc.Messages;
61 import net.sf.hajdbc.SQLException;
62 import net.sf.hajdbc.SynchronizationStrategy;
63 import net.sf.hajdbc.SynchronizationStrategyBuilder;
64 import net.sf.hajdbc.sql.DataSourceDatabase;
65 import net.sf.hajdbc.sql.DriverDatabase;
66 import net.sf.hajdbc.util.concurrent.CronThreadPoolExecutor;
67 import net.sf.hajdbc.util.concurrent.SynchronousExecutor;
68
69 import org.slf4j.Logger;
70 import org.slf4j.LoggerFactory;
71
72
77 public class LocalDatabaseCluster implements DatabaseCluster
78 {
79 private static final String STATE_DELIMITER = ",";
80
81 private static Preferences preferences = Preferences.userNodeForPackage(LocalDatabaseCluster.class);
82 static Logger logger = LoggerFactory.getLogger(LocalDatabaseCluster.class);
83
84 private String id;
85 private Balancer balancer;
86 private Dialect dialect;
87 private String defaultSynchronizationStrategyId;
88 private String failureDetectionSchedule;
89 private String autoActivationSchedule;
90 private int minThreads;
91 private int maxThreads;
92 private int maxIdle;
93 private Transaction transaction;
94
95 private Map <String , Database> databaseMap = new ConcurrentHashMap <String , Database>();
96 private Map <Database, Object > connectionFactoryMap = new ConcurrentHashMap <Database, Object >();
97 private ExecutorService transactionalExecutor;
98 private ExecutorService nonTransactionalExecutor;
99 private CronThreadPoolExecutor cronExecutor = new CronThreadPoolExecutor(2);
100 private ReadWriteLock lock = createReadWriteLock();
101
102
106 private static ReadWriteLock createReadWriteLock()
107 {
108 try
109 {
110 return new ReentrantReadWriteLock (true);
111 }
112 catch (NoSuchMethodError e)
113 {
114 return new ReentrantReadWriteLock ();
115 }
116 }
117
118
121 public String [] loadState() throws java.sql.SQLException
122 {
123 try
124 {
125 preferences.sync();
126
127 String state = preferences.get(this.id, null);
128
129 if (state == null)
130 {
131 return null;
132 }
133
134 if (state.length() == 0)
135 {
136 return new String [0];
137 }
138
139 String [] databases = state.split(STATE_DELIMITER);
140
141 for (String id: databases)
143 {
144 if (!this.databaseMap.containsKey(id))
145 {
146 preferences.remove(this.id);
148 preferences.flush();
149
150 return null;
151 }
152 }
153
154 return databases;
155 }
156 catch (BackingStoreException e)
157 {
158 throw new SQLException(Messages.getMessage(Messages.CLUSTER_STATE_LOAD_FAILED, this), e);
159 }
160 }
161
162
165 public Map <Database, ?> getConnectionFactoryMap()
166 {
167 return this.connectionFactoryMap;
168 }
169
170
173 public boolean isAlive(Database database)
174 {
175 Connection connection = null;
176
177 try
178 {
179 connection = database.connect(this.connectionFactoryMap.get(database));
180
181 Statement statement = connection.createStatement();
182
183 statement.execute(this.dialect.getSimpleSQL());
184
185 statement.close();
186
187 return true;
188 }
189 catch (java.sql.SQLException e)
190 {
191 return false;
192 }
193 finally
194 {
195 if (connection != null)
196 {
197 try
198 {
199 connection.close();
200 }
201 catch (java.sql.SQLException e)
202 {
203 logger.warn(e.toString(), e);
204 }
205 }
206 }
207 }
208
209
212 public synchronized boolean deactivate(Database database)
213 {
214 MBeanServer server = DatabaseClusterFactory.getMBeanServer();
215
216 try
217 {
218 ObjectName name = DatabaseClusterFactory.getObjectName(this.id, database.getId());
219
220 if (server.isRegistered(name))
222 {
223 server.unregisterMBean(name);
224 }
225
226 server.registerMBean(new StandardMBean (database, database.getInactiveMBeanClass()), name);
227 }
228 catch (JMException e)
229 {
230 throw new IllegalStateException (e.toString(), e);
231 }
232
233 boolean removed = this.balancer.remove(database);
234
235 if (removed)
236 {
237 this.storeState();
238 }
239
240 return removed;
241 }
242
243
246 public String getId()
247 {
248 return this.id;
249 }
250
251
254 public synchronized boolean activate(Database database)
255 {
256 MBeanServer server = DatabaseClusterFactory.getMBeanServer();
257
258 try
259 {
260 ObjectName name = DatabaseClusterFactory.getObjectName(this.id, database.getId());
261
262 if (server.isRegistered(name))
264 {
265 server.unregisterMBean(name);
266 }
267
268 server.registerMBean(new StandardMBean (database, database.getActiveMBeanClass()), name);
269
270 if (database.isDirty())
271 {
272 DatabaseClusterFactory.getInstance().exportConfiguration();
273
274 database.clean();
275 }
276
277 boolean added = this.balancer.add(database);
278
279 if (added)
280 {
281 this.storeState();
282 }
283
284 return added;
285 }
286 catch (JMException e)
287 {
288 throw new IllegalStateException (e);
289 }
290 }
291
292 private void storeState()
293 {
294 StringBuilder builder = new StringBuilder ();
295
296 Iterator <Database> databases = this.balancer.list().iterator();
297
298 while (databases.hasNext())
299 {
300 builder.append(databases.next().getId());
301
302 if (databases.hasNext())
303 {
304 builder.append(STATE_DELIMITER);
305 }
306 }
307
308 preferences.put(this.id, builder.toString());
309
310 try
311 {
312 preferences.flush();
313 }
314 catch (BackingStoreException e)
315 {
316 logger.warn(Messages.getMessage(Messages.CLUSTER_STATE_STORE_FAILED, this), e);
317 }
318 }
319
320
323 public Collection <String > getActiveDatabases()
324 {
325 return this.extractIdentifiers(this.balancer.list());
326 }
327
328
331 public Collection <String > getInactiveDatabases()
332 {
333 return this.extractIdentifiers(this.getInactiveDatabaseSet());
334 }
335
336 protected Set <Database> getInactiveDatabaseSet()
337 {
338 Set <Database> databaseSet = new TreeSet <Database>(this.databaseMap.values());
339
340 databaseSet.removeAll(this.balancer.list());
341
342 return databaseSet;
343 }
344
345 private List <String > extractIdentifiers(Collection <Database> databases)
346 {
347 List <String > databaseList = new ArrayList <String >(databases.size());
348
349 for (Database database: databases)
350 {
351 databaseList.add(database.getId());
352 }
353
354 return databaseList;
355 }
356
357
360 public Database getDatabase(String id)
361 {
362 Database database = this.databaseMap.get(id);
363
364 if (database == null)
365 {
366 throw new IllegalArgumentException (Messages.getMessage(Messages.INVALID_DATABASE, id, this));
367 }
368
369 return database;
370 }
371
372
375 public SynchronizationStrategy getDefaultSynchronizationStrategy()
376 {
377 return DatabaseClusterFactory.getInstance().getSynchronizationStrategy(this.defaultSynchronizationStrategyId);
378 }
379
380
383 public Balancer getBalancer()
384 {
385 return this.balancer;
386 }
387
388
391 public ExecutorService getTransactionalExecutor()
392 {
393 return this.transactionalExecutor;
394 }
395
396
399 public ExecutorService getNonTransactionalExecutor()
400 {
401 return this.nonTransactionalExecutor;
402 }
403
404
407 public Dialect getDialect()
408 {
409 return this.dialect;
410 }
411
412
415 public Lock readLock()
416 {
417 return this.lock.readLock();
418 }
419
420
423 public Lock writeLock()
424 {
425 return this.lock.writeLock();
426 }
427
428
431 public final boolean isAlive(String id)
432 {
433 return this.isAlive(this.getDatabase(id));
434 }
435
436
439 public final void deactivate(String databaseId)
440 {
441 if (this.deactivate(this.getDatabase(databaseId)))
442 {
443 logger.info(Messages.getMessage(Messages.DATABASE_DEACTIVATED, databaseId, this));
444 }
445 }
446
447
450 public final void activate(String databaseId)
451 {
452 this.activate(databaseId, this.getDefaultSynchronizationStrategy());
453 }
454
455
458 public final void activate(String databaseId, String strategyId)
459 {
460 this.activate(databaseId, DatabaseClusterFactory.getInstance().getSynchronizationStrategy(strategyId));
461 }
462
463
466 public String getVersion()
467 {
468 return DatabaseClusterFactory.getVersion();
469 }
470
471
478 public final void handleFailure(Database database, java.sql.SQLException cause) throws java.sql.SQLException
479 {
480 if (this.isAlive(database))
481 {
482 throw cause;
483 }
484
485 if (this.deactivate(database))
486 {
487 logger.warn(Messages.getMessage(Messages.DATABASE_NOT_ALIVE, database, this), cause);
488 }
489 }
490
491
494 public void add(String id, String driver, String url)
495 {
496 DriverDatabase database = new DriverDatabase();
497
498 database.setId(id);
499 database.setDriver(driver);
500 database.setUrl(url);
501
502 this.register(database);
503
504 this.add(database);
505 }
506
507
510 public void add(String id, String name)
511 {
512 DataSourceDatabase database = new DataSourceDatabase();
513
514 database.setId(id);
515 database.setName(name);
516
517 this.register(database);
518
519 this.add(database);
520 }
521
522 private void register(Database database)
523 {
524 MBeanServer server = DatabaseClusterFactory.getMBeanServer();
525
526 try
527 {
528 ObjectName name = DatabaseClusterFactory.getObjectName(this.id, database.getId());
529
530 server.registerMBean(new StandardMBean (database, database.getInactiveMBeanClass()), name);
531 }
532 catch (JMException e)
533 {
534 logger.error(e.toString(), e);
535
536 throw new IllegalStateException (e);
537 }
538 }
539
540
543 public synchronized void remove(String id)
544 {
545 Database database = this.getDatabase(id);
546
547 if (this.balancer.contains(database))
548 {
549 throw new IllegalStateException (Messages.getMessage(Messages.DATABASE_STILL_ACTIVE, id, this));
550 }
551
552 this.unregister(database);
553
554 this.databaseMap.remove(id);
555 this.connectionFactoryMap.remove(database);
556
557 DatabaseClusterFactory.getInstance().exportConfiguration();
558 }
559
560 private void unregister(Database database)
561 {
562 MBeanServer server = DatabaseClusterFactory.getMBeanServer();
563
564 try
565 {
566 ObjectName name = DatabaseClusterFactory.getObjectName(this.id, database.getId());
567
568 server.unregisterMBean(name);
569 }
570 catch (JMException e)
571 {
572 logger.error(e.toString(), e);
573
574 throw new IllegalStateException (e);
575 }
576 }
577
578
581 public void start() throws java.sql.SQLException
582 {
583 MBeanServer server = DatabaseClusterFactory.getMBeanServer();
584
585 try
586 {
587 server.registerMBean(new StandardMBean (this, DatabaseClusterMBean.class), DatabaseClusterFactory.getObjectName(this.id));
588
589 for (Database database: this.databaseMap.values())
590 {
591 ObjectName name = DatabaseClusterFactory.getObjectName(this.id, database.getId());
592
593 server.registerMBean(new StandardMBean (database, database.getInactiveMBeanClass()), name);
594 }
595 }
596 catch (JMException e)
597 {
598 throw new SQLException(e);
599 }
600
601 String [] databases = this.loadState();
602
603 if (databases != null)
604 {
605 for (String id: databases)
606 {
607 this.activate(this.getDatabase(id));
608 }
609 }
610 else
611 {
612 for (String id: this.getInactiveDatabases())
613 {
614 Database database = this.getDatabase(id);
615
616 if (this.isAlive(database))
617 {
618 this.activate(database);
619 }
620 }
621 }
622
623 this.nonTransactionalExecutor = new ThreadPoolExecutor (this.minThreads, this.maxThreads, this.maxIdle, TimeUnit.SECONDS, new SynchronousQueue <Runnable >(), new ThreadPoolExecutor.CallerRunsPolicy ());
624
625 this.transactionalExecutor = this.transaction.equals(Transaction.XA) ? new SynchronousExecutor() : this.nonTransactionalExecutor;
626
627 if (this.failureDetectionSchedule != null)
628 {
629 this.cronExecutor.schedule(new FailureDetectionTask(), this.failureDetectionSchedule);
630 }
631
632 if (this.autoActivationSchedule != null)
633 {
634 this.cronExecutor.schedule(new AutoActivationTask(), this.autoActivationSchedule);
635 }
636 }
637
638
641 public synchronized void stop()
642 {
643 MBeanServer server = DatabaseClusterFactory.getMBeanServer();
644
645 for (String databaseId: this.databaseMap.keySet())
646 {
647 try
648 {
649 ObjectName name = DatabaseClusterFactory.getObjectName(this.id, databaseId);
650
651 if (server.isRegistered(name))
652 {
653 server.unregisterMBean(name);
654 }
655 }
656 catch (JMException e)
657 {
658 logger.warn(e.getMessage(), e);
659 }
660 }
661
662 try
663 {
664 ObjectName name = DatabaseClusterFactory.getObjectName(this.id);
665
666 if (server.isRegistered(name))
667 {
668 server.unregisterMBean(name);
669 }
670 }
671 catch (JMException e)
672 {
673 logger.warn(e.getMessage(), e);
674 }
675
676 this.cronExecutor.shutdownNow();
677
678 if (this.nonTransactionalExecutor != null)
679 {
680 this.nonTransactionalExecutor.shutdownNow();
681 }
682
683 if (this.transactionalExecutor != null)
684 {
685 this.transactionalExecutor.shutdownNow();
686 }
687 }
688
689
692 @Override
693 public final String toString()
694 {
695 return this.getId();
696 }
697
698
701 @Override
702 public final boolean equals(Object object)
703 {
704 DatabaseCluster databaseCluster = (DatabaseCluster) object;
705
706 return this.getId().equals(databaseCluster.getId());
707 }
708
709 SynchronizationStrategyBuilder getDefaultSynchronizationStrategyBuilder()
710 {
711 return new SynchronizationStrategyBuilder(this.defaultSynchronizationStrategyId);
712 }
713
714 void setDefaultSynchronizationStrategyBuilder(SynchronizationStrategyBuilder builder)
715 {
716 this.defaultSynchronizationStrategyId = builder.getId();
717 }
718
719 synchronized void add(Database database)
720 {
721 String id = database.getId();
722
723 if (this.databaseMap.containsKey(id))
724 {
725 throw new IllegalArgumentException (Messages.getMessage(Messages.DATABASE_ALREADY_EXISTS, id, this));
726 }
727
728 this.connectionFactoryMap.put(database, database.createConnectionFactory());
729 this.databaseMap.put(id, database);
730 }
731
732 Iterator <Database> getDriverDatabases()
733 {
734 return this.getDatabases(Driver .class);
735 }
736
737 Iterator <Database> getDataSourceDatabases()
738 {
739 return this.getDatabases(DataSource .class);
740 }
741
742 Iterator <Database> getDatabases(Class targetClass)
743 {
744 List <Database> databaseList = new ArrayList <Database>(this.databaseMap.size());
745
746 for (Database database: this.databaseMap.values())
747 {
748 if (targetClass.equals(database.getConnectionFactoryClass()))
749 {
750 databaseList.add(database);
751 }
752 }
753
754 return databaseList.iterator();
755 }
756
757 private void activate(String databaseId, SynchronizationStrategy strategy)
758 {
759 try
760 {
761 if (this.activate(this.getDatabase(databaseId), strategy))
762 {
763 logger.info(Messages.getMessage(Messages.DATABASE_ACTIVATED, databaseId, this));
764 }
765 }
766 catch (java.sql.SQLException e)
767 {
768 logger.error(Messages.getMessage(Messages.DATABASE_ACTIVATE_FAILED, databaseId, this), e);
769
770 java.sql.SQLException exception = e.getNextException();
771
772 while (exception != null)
773 {
774 logger.error(exception.getMessage(), e);
775
776 exception = exception.getNextException();
777 }
778
779 throw new IllegalStateException (e.toString());
780 }
781 catch (InterruptedException e)
782 {
783 logger.warn(e.toString(), e);
784 throw new IllegalMonitorStateException (e.toString());
785 }
786 }
787
788 boolean activate(Database database, SynchronizationStrategy strategy) throws java.sql.SQLException , InterruptedException
789 {
790 if (this.getBalancer().contains(database))
791 {
792 return false;
793 }
794
795 if (!this.isAlive(database))
796 {
797 return false;
798 }
799
800 Lock lock = this.writeLock();
801
802 lock.lockInterruptibly();
803
804 try
805 {
806 List <Database> databaseList = this.getBalancer().list();
807
808 if (databaseList.isEmpty())
809 {
810 return this.activate(database);
811 }
812
813 this.activate(database, databaseList, strategy);
814
815 return true;
816 }
817 finally
818 {
819 lock.unlock();
820 }
821 }
822
823 private void activate(Database inactiveDatabase, List <Database> activeDatabaseList, SynchronizationStrategy strategy) throws java.sql.SQLException
824 {
825 Database activeDatabase = this.getBalancer().next();
826
827 Connection inactiveConnection = null;
828 Connection activeConnection = null;
829
830 List <Connection > connectionList = new ArrayList <Connection >(activeDatabaseList.size());
831
832 try
833 {
834 Map <Database, ?> connectionFactoryMap = this.getConnectionFactoryMap();
835
836 inactiveConnection = inactiveDatabase.connect(connectionFactoryMap.get(inactiveDatabase));
837
838 Map <String , List <String >> schemaMap = new HashMap <String , List <String >>();
839
840 DatabaseMetaData metaData = inactiveConnection.getMetaData();
841
842 ResultSet resultSet = metaData.getTables(null, null, "%", new String [] { "TABLE" });
843
844 while (resultSet.next())
845 {
846 String table = resultSet.getString("TABLE_NAME");
847 String schema = resultSet.getString("TABLE_SCHEM");
848
849 List <String > tableList = schemaMap.get(schema);
850
851 if (tableList == null)
852 {
853 tableList = new LinkedList <String >();
854
855 schemaMap.put(schema, tableList);
856 }
857
858 tableList.add(table);
859 }
860
861 resultSet.close();
862
863 activeConnection = activeDatabase.connect(connectionFactoryMap.get(activeDatabase));
864
865 Dialect dialect = this.getDialect();
866
867 if (strategy.requiresTableLocking())
868 {
869 logger.info(Messages.getMessage(Messages.TABLE_LOCK_ACQUIRE));
870
871 Map <String , Map <String , String >> lockTableSQLMap = new HashMap <String , Map <String , String >>();
872
873 for (Database database: activeDatabaseList)
875 {
876 Connection connection = database.equals(activeDatabase) ? activeConnection : database.connect(connectionFactoryMap.get(database));
877
878 connectionList.add(connection);
879
880 connection.setAutoCommit(false);
881 connection.setTransactionIsolation(Connection.TRANSACTION_READ_COMMITTED);
882
883 Statement statement = connection.createStatement();
884
885 for (Map.Entry <String , List <String >> schemaMapEntry: schemaMap.entrySet())
886 {
887 String schema = schemaMapEntry.getKey();
888
889 Map <String , String > map = lockTableSQLMap.get(schema);
890
891 if (map == null)
892 {
893 map = new HashMap <String , String >();
894
895 lockTableSQLMap.put(schema, map);
896 }
897
898 for (String table: schemaMapEntry.getValue())
899 {
900 String sql = map.get(table);
901
902 if (sql == null)
903 {
904 sql = dialect.getLockTableSQL(metaData, schema, table);
905
906 logger.debug(sql);
907
908 map.put(table, sql);
909 }
910
911 statement.execute(sql);
912 }
913 }
914
915 statement.close();
916 }
917 }
918
919 logger.info(Messages.getMessage(Messages.DATABASE_SYNC_START, inactiveDatabase, this));
920
921 strategy.synchronize(inactiveConnection, activeConnection, schemaMap, dialect);
922
923 logger.info(Messages.getMessage(Messages.DATABASE_SYNC_END, inactiveDatabase, this));
924
925 this.activate(inactiveDatabase);
926
927 if (strategy.requiresTableLocking())
928 {
929 logger.info(Messages.getMessage(Messages.TABLE_LOCK_ACQUIRE));
930
931 this.rollback(connectionList);
933 }
934 }
935 catch (java.sql.SQLException e)
936 {
937 this.rollback(connectionList);
938
939 throw e;
940 }
941 finally
942 {
943 this.close(activeConnection);
944 this.close(inactiveConnection);
945
946 for (Connection connection: connectionList)
947 {
948 this.close(connection);
949 }
950 }
951 }
952
953 private void rollback(List <Connection > connectionList)
954 {
955 for (Connection connection: connectionList)
956 {
957 try
958 {
959 connection.rollback();
960 connection.setAutoCommit(true);
961 }
962 catch (java.sql.SQLException e)
963 {
964 logger.warn(e.toString(), e);
965 }
966 }
967 }
968
969 private void close(Connection connection)
970 {
971 if (connection != null)
972 {
973 try
974 {
975 if (!connection.isClosed())
976 {
977 connection.close();
978 }
979 }
980 catch (java.sql.SQLException e)
981 {
982 logger.warn(e.toString(), e);
983 }
984 }
985 }
986
987 private class FailureDetectionTask implements Runnable
988 {
989
992 public void run()
993 {
994 for (Database database: LocalDatabaseCluster.this.getBalancer().list())
995 {
996 if (!LocalDatabaseCluster.this.isAlive(database))
997 {
998 if (LocalDatabaseCluster.this.deactivate(database))
999 {
1000 logger.warn(Messages.getMessage(Messages.DATABASE_NOT_ALIVE, database, LocalDatabaseCluster.this));
1001 }
1002 }
1003 }
1004 }
1005 }
1006
1007 private class AutoActivationTask implements Runnable
1008 {
1009
1012 public void run()
1013 {
1014 for (Database database: LocalDatabaseCluster.this.getInactiveDatabaseSet())
1015 {
1016 try
1017 {
1018 if (LocalDatabaseCluster.this.activate(database, LocalDatabaseCluster.this.getDefaultSynchronizationStrategy()))
1019 {
1020 logger.info(Messages.getMessage(Messages.DATABASE_ACTIVATED, database, LocalDatabaseCluster.this));
1021 }
1022 }
1023 catch (java.sql.SQLException e)
1024 {
1025 logger.warn(Messages.getMessage(Messages.DATABASE_ACTIVATE_FAILED, database, LocalDatabaseCluster.this), e);
1026
1027 java.sql.SQLException exception = e.getNextException();
1028
1029 while (exception != null)
1030 {
1031 logger.warn(exception.getMessage(), e);
1032
1033 exception = exception.getNextException();
1034 }
1035 }
1036 catch (InterruptedException e)
1037 {
1038 break;
1039 }
1040 }
1041 }
1042 }
1043}
1044 |
Popular Tags |
Java API By Example, From Geeks To Geeks.