Skip to content
This repository was archived by the owner on Jun 2, 2021. It is now read-only.

DelayQueue implementation, statuses and complex object serialization. #1

Open
Matzz wants to merge 33 commits into
base: master
Choose a base branch
from
Open

DelayQueue implementation, statuses and complex object serialization. #1

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
33 commits
Select commit Hold shift + click to select a range
47bb1e3
Delayed queue, not tested yet
Jul 14, 2014
752e45c
formatting fix
Jul 14, 2014
32aef99
formatting fix2
Jul 14, 2014
fb128c1
added some test + fixes
Jul 14, 2014
6725ea2
major refactor
Jul 15, 2014
ca160c3
test database config in separate file
Jul 15, 2014
b7cc060
test fixes
Jul 15, 2014
ed8988f
gitignore
Jul 15, 2014
2b8042e
Delayed queue fixes + tests
Jul 16, 2014
d8f8afe
Statusable queues
Jul 21, 2014
b53c218
Update README.md
Matzz Jul 21, 2014
cd0bc26
Statuses doc
Matzz Jul 21, 2014
67a8a76
take with parameters
Jul 24, 2014
ba0977d
Merge branch 'master' of https://github.com/Matzz/db-patterns
Jul 24, 2014
fa1cd96
code formatting
Jul 29, 2014
bbe62d0
standalone version
Jul 29, 2014
d137f28
v3.1
Jul 29, 2014
4ffc2b5
mysql-connector dep
Jul 29, 2014
79ebf9c
configurable take blocking time
Jul 30, 2014
60dfc00
Merge remote-tracking branch 'origin/master' into standalone
Jul 30, 2014
545c2d4
0.3.3
Jul 30, 2014
c64952a
Added priorities
Aug 12, 2014
6e6c57b
Reverted pom
Aug 12, 2014
32373bb
Internally inverted priorities to enable sort index.
Sep 18, 2014
3e7a70a
Indexes fix
Sep 22, 2014
6846eb2
Enhanced queue pull performance.
Sep 24, 2014
d6069f1
Merge remote-tracking branch 'origin/standalone'
Sep 24, 2014
f4c4118
Handling deadlocks
Sep 25, 2014
edf7aec
Rewrited sql to prevend deadlocks
Sep 25, 2014
d88270a
version 0.3.9
Sep 25, 2014
90bb751
Fixed leaking connections in delayed Queue. Added value uniqueness test
Sep 25, 2014
8c4c9fc
Merge remote-tracking branch 'origin/standalone'
Sep 25, 2014
c4d17e6
Removed println
Sep 25, 2014
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 27 additions & 0 deletions .gitignore
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,30 @@
*target*
*.jar
*.war
*.ear
*.class

# eclipse specific git ignore
*.pydevproject
.project
.metadata
bin/**
tmp/**
tmp/**/*
*.tmp
*.bak
*.swp
*~.nib
local.properties
.classpath
.settings/
.loadpath

# External tool builders
.externalToolBuilders/

# Locally stored "Eclipse launch configurations"
*.launch
target/
pom.xml.tag
pom.xml.releaseBackup
Expand Down
58 changes: 55 additions & 3 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
MySQL DB Patterns
=================
by [Andrew Brampton](http://bramp.net) 2013
by [Andrew Brampton](http://bramp.net) 2013, contributed by [Mateusz Zakarczemny](https://github.com/Matzz)

Intro
-----
Expand Down Expand Up @@ -36,14 +36,16 @@ The MySQLSleepBasedCondition is based on the MySQL ``SLEEP()`` and ``KILL QUERY`
The thread that is woken up is guaranteed to be the one that has waited the longest.


Queue
Blocking queue
-----

A distributed MySQL backed Java BlockingQueue

```java
DataSource ds = ...
BlockingQueue<String> queue = new MySQLBasedQueue<String>(ds, "queue name", String.class);

//datasoruce, queue table, queuename, value type, thread name
BlockingQueue<String> queue = new MySQLBasedQueue<String>(ds, "queue", "queue name", String.class, "Worker1");
queue.add("Some String");

// on another thread (or process, or machine)
Expand All @@ -55,6 +57,56 @@ A distributed MySQL backed Java BlockingQueue
The MySQLBasedQueue uses the MySQLSleepBasedCondition to help form a blocking
queue, that can work without polling the database for new work.

More complex types could be stored using serializator:
```java
Serializator serializator = new DefaultSerializator<MyType>();
BlockingQueue<String> queue = new MySQLBasedQueue<String>(ds, "queue", "queue name", serializator, "Worker1");
MyType value = new MyType(...);
queue.add(value);
```
DefaultSerializator serializes values using java ObjectOutputStream but other implementation might be passed to queue (eg. some custom JsonSerializer).

DelayQueue
-----------------
A distributed MySQL backed Java DelayQueue

```java
Serializator serializator = new DefaultSerializator<MyDelayedType>(); // MyType must extends Delayed interface
DelayQueue<String> queue = new MySQLBasedDelayQueue<String>(ds, "queue", "queue name", serializator, "Worker1");
MyDelayedType value = new MyDelayedType(10, TimeUnit.SECONDS);
queue.add(value);
queue.peek(); // equals null
Thread.sleep(11*1000);
queue.peek(); // equals value

```

Statuses
-----------------
MySQLBasedQueue, MySQLBasedDelayQueue implements StatusableQueue interface which enables setting queue item statuses. Statuses do not affect polling of items. They might be set at any time and to any value. They just provides convenient way of tracking item state.

PriorityQueue brings such methods:
```java
public V pollWithMetadata();
public V pollWithMetadata(long timeout, TimeUnit unit) throws InterruptedException;
public V peekWithMetadata();
public void updateStatus(long id, String newStatus);
public String getStatus(long id);
```
Where V is class implementing ValueWithMetadata interface. ValueWithMetadata contains item id in queue, status and item value.


Priority
-----------------
MySQLBasedQueue, MySQLBasedDelayQueue implements PriorityQueue interface which enables setting items priority. The higher priority is, the earlier item will be polled from queue.
From 0.3.5 version, due to lack of DESC index in mysql, internally in table values are stored inverted. While migrating to version 0.3.5 all priorities should be multiplied by -1.

StatusableQueue brings such methods:
```java
public boolean add(E value, int priority);
```
Default priority for add is 0. Priority could be retrieved from value metadata (see above) using getPriority method.


Build and Release
-----------------
Expand Down
223 changes: 110 additions & 113 deletions pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,119 +1,116 @@
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>

<groupId>net.bramp.db-patterns</groupId>
<artifactId>db-patterns</artifactId>
<version>0.2-SNAPSHOT</version>
<packaging>jar</packaging>

<name>DB Patterns</name>
<description>Some simple DB patterns implemented onto of MySQL</description>
<url>https://github.com/bramp/db-patterns</url>

<developers>
<developer>
<id>bramp</id>
<name>Andrew Brampton</name>
</developer>
</developers>

<licenses>
<license>
<name>The BSD 2-Clause License</name>
<url>http://opensource.org/licenses/BSD-2-Clause</url>
<distribution>repo</distribution>
</license>
</licenses>

<scm>
<url>https://github.com/bramp/db-patterns</url>
<connection>scm:git:git@github.com:bramp/db-patterns.git</connection>
<tag>HEAD</tag>
<modelVersion>4.0.0</modelVersion>

<groupId>net.bramp.db-patterns</groupId>
<artifactId>db-patterns</artifactId>
<version>0.2-SNAPSHOT</version>
<packaging>jar</packaging>

<name>DB Patterns</name>
<description>Some simple DB patterns implemented onto of MySQL</description>
<url>https://github.com/bramp/db-patterns</url>

<developers>
<developer>
<id>bramp</id>
<name>Andrew Brampton</name>
</developer>
</developers>

<licenses>
<license>
<name>The BSD 2-Clause License</name>
<url>http://opensource.org/licenses/BSD-2-Clause</url>
<distribution>repo</distribution>
</license>
</licenses>

<scm>
<url>https://github.com/bramp/db-patterns</url>
<connection>scm:git:git@github.com:bramp/db-patterns.git</connection>
<tag>HEAD</tag>
</scm>

<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>

<logback.version>0.9.30</logback.version>
</properties>

<dependencies>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>1.6.2</version>
</dependency>
<dependency>
<groupId>com.google.code.findbugs</groupId>
<artifactId>jsr305</artifactId>
<version>2.0.2</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>net.sourceforge.findbugs</groupId>
<artifactId>annotations</artifactId>
<version>1.3.2</version>
<scope>provided</scope>
</dependency>

<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.27</version>
<scope>provided</scope>
</dependency>

<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.11</version>
<scope>test</scope>
</dependency>

<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-core</artifactId>
<version>${logback.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
<version>${logback.version}</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
<version>15.0</version>
<scope>test</scope>
</dependency>
</dependencies>

<distributionManagement>
<snapshotRepository>
<id>ossrh</id>
<url>https://oss.sonatype.org/content/repositories/snapshots</url>
</snapshotRepository>
<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>

<logback.version>0.9.30</logback.version>
</properties>

<dependencies>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>1.6.2</version>
</dependency>

<dependency>
<groupId>com.google.code.findbugs</groupId>
<artifactId>jsr305</artifactId>
<version>2.0.2</version>
<scope>provided</scope>
</dependency>

<dependency>
<groupId>net.sourceforge.findbugs</groupId>
<artifactId>annotations</artifactId>
<version>1.3.2</version>
<scope>provided</scope>
</dependency>

<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.27</version>
<scope>provided</scope>
</dependency>

<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.11</version>
<scope>test</scope>
</dependency>

<dependency>
<groupId>ch.qos.logback</groupId>
<artifactId>logback-classic</artifactId>
<version>${logback.version}</version>
<scope>test</scope>
</dependency>

<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
<version>15.0</version>
<scope>test</scope>
</dependency>
</dependencies>

<distributionManagement>
<snapshotRepository>
<id>ossrh</id>
<url>https://oss.sonatype.org/content/repositories/snapshots</url>
</snapshotRepository>
<repository>
<id>ossrh</id>
<url>https://oss.sonatype.org/service/local/staging/deploy/maven2/</url>
</repository>
</distributionManagement>

<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.1</version>
<configuration>
<source>1.6</source>
<target>1.6</target>
<encoding>UTF-8</encoding>
</configuration>
</plugin>
</distributionManagement>

<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.1</version>
<configuration>
<source>1.6</source>
<target>1.6</target>
<encoding>UTF-8</encoding>
</configuration>
</plugin>

<plugin>
<groupId>org.apache.maven.plugins</groupId>
Expand Down Expand Up @@ -201,8 +198,8 @@
</configuration>
</plugin>

</plugins>
</plugins>

<defaultGoal>install</defaultGoal>
</build>
</project>
<defaultGoal>install</defaultGoal>
</build>
</project>
Loading