1
\$\begingroup\$

The app suppose to solve a subset of Producer–consumer problem. One producer, one consumer thread and buffer. The idea of the app to handle correctly database downtimes. The solution is to keep producing timestamps in ArrayBlockingQueue. When it is reached threshold (only a few empty slots remaining) the Buffer will serialize ArrayBlockingQueue to file and keep track in LIFO queue all filenames. In case the allowed amount of files created the Producer will be halted by RuntimeException.

Would be great to have feedback on overall design, testability and general advice on how to improve further. I have doubts about choosing the right approach while testing a temporary database outage.

Producer:

public class Producer extends CustomThread implements Runnable{
    private Buffer buffer;

    public Producer(String name, Buffer buffer) {
        super(name);
        this.buffer = buffer;
    }

    @Override
    public void run() {
        Thread.currentThread().setName(threadName);
        Timestamp timestamp = Timestamp.from(Instant.now());
        log.info("Produced {}", timestamp);
        try {
            buffer.put(timestamp);
        } catch (Exception e) {
            log.info("Stopping Producer");
            log.error(e.getMessage());
            throw new RuntimeException(e.getMessage());
        }
    }
}

Consumer:

public class Consumer extends CustomThread implements Runnable{

    private Buffer buffer;
    private TimestampService timestampService;

    public Consumer(String name, Buffer buffer, TimestampService timestampService) {
        super(name);
        this.buffer = buffer;
        this.timestampService = timestampService;
    }

    @Override
    public void run() {
        Thread.currentThread().setName(threadName);
        List<Timestamp> batch = buffer.getBatch();

        while (!batch.isEmpty()) {
            log.info("Consumed: {}", batch);
            try {
                timestampService.insertBatch(batch);
            } catch (InterruptedException e) {
                log.debug(e.getMessage());
            }
            batch = buffer.getBatch();
        }
    }
}

FileCache (de/serializes Buffer)

public class FileCache {

    @Getter
    private Queue<String> fileQueue = new LinkedList<>();
    private static final int QUEUE_MAX_SIZE = 0;

    public void flushToDisk(List<Timestamp> timestamps) {
        if(fileQueue.size() + 1 > QUEUE_MAX_SIZE) {
            throw new RuntimeException("Disk quota exhausted when trying to flush buffer to disk");
        }

        FileOutputStream fileOutputStream;
        ObjectOutputStream objectOutputStream = null;
        try {
            String fileName = String.format("%s-timestamps.txt", timestamps.get(0).toInstant().toString().replaceAll(":", ""));
            fileOutputStream = new FileOutputStream(fileName);
            objectOutputStream = new ObjectOutputStream(fileOutputStream);
            objectOutputStream.writeObject(timestamps);
            objectOutputStream.flush();
            objectOutputStream.close();
            fileQueue.add(fileName);
            log.info("Successfully persisted buffer to file: {}. File queue size: {}", fileName, fileQueue.size());
        } catch (IOException e) {
           log.info(e.getMessage());
           throw new RuntimeException("Exception while flushing buffer to file");
        }
    }

    @SuppressWarnings("unchecked")
    public List<Timestamp> getTimestamps() {
        List<Timestamp> timestamps = new ArrayList<>();
        String fileName = fileQueue.poll();

        if(fileName != null) {
            try {
                FileInputStream fileInputStream = new FileInputStream(fileName);
                ObjectInputStream objectInputStream = new ObjectInputStream(fileInputStream);
                timestamps = (List<Timestamp>) objectInputStream.readObject();
                objectInputStream.close();
                log.info("Successfully retrieved one snapshot of latest buffer from file: " + fileName);
            } catch (IOException | ClassNotFoundException e) {
                log.info(e.getMessage());
            }
        }
        return timestamps;
    }
}

Service is responsible for inserting batches of timestamps to mysql, getting all records, deleting

public class TimestampService {

    private Jdbi jdbi;

    public TimestampService() {
        DataSource ds = db.DataSource.getDs();
        jdbi = Jdbi.create(ds);
        jdbi.installPlugin(new SqlObjectPlugin());
    }

    public void insertBatch(List<Timestamp> batch) throws InterruptedException {
        log.info("Inserting batch: {}", batch);
        int[] result = {0};
        do {
            try {
                result = jdbi.withExtension(TimestampDao.class, dao -> dao.bulkInsert(batch));
            } catch (Exception e) {
                log.error(e.getMessage());
            }
            if(!isBatchInserted(result)) {
                log.info("Batch insert failed. Retrying in 5 sec");
                TimeUnit.SECONDS.sleep(5);
            } else {
                log.info("Batch inserted successfully: {}", batch);
            }
        } while(!isBatchInserted(result));
    }

    /**
     * https://docs.oracle.com/javase/7/docs/api/java/sql/Statement.html#SUCCESS_NO_INFO
     * @param result
     * @return
     */
    private boolean isBatchInserted(int[] result) {
        return !IntStream.range(0, result.length).allMatch(i -> i == Statement.SUCCESS_NO_INFO);
    }

    public List<Timestamp> findAllTimestamps() {
        log.info("Finding all timestamps");
        return jdbi.withExtension(TimestampDao.class, TimestampDao::findAllTimestamps);
    };

    public boolean deleteAllTimestamps() {
        log.info("Deleting all timestamps");
        return jdbi.withExtension(TimestampDao.class, TimestampDao::deleteAllTimestamps);
    };
}

DataSource that is singleton, initializes HikariCP DataSource impl that is later passed to jdbi lib to perform batch inserts.

public class DataSource {

    private static final String CONFIG_FILE = "src/main/resources/db/db.properties";
    private static HikariConfig config = new HikariConfig(CONFIG_FILE);

    private static HikariDataSource ds;

    static{
        try {
            ds = new HikariDataSource(config);
        } catch(CJCommunicationsException | HikariPool.PoolInitializationException e) {
            log.info(e.getMessage());
        }
    }

    static class Holder {
        static DataSource INSTANCE = new DataSource();
    }

    private DataSource() { }

    public static DataSource getInstance() {
        return Holder.INSTANCE;
    }

    public static Connection getConnection() throws SQLException{
            return ds.getConnection();
    }

    public static HikariDataSource getDs() {
        return ds;
    }
}

Dao impl of jdbi SQL objects

public interface TimestampDao {

    @SqlBatch("INSERT INTO test values (:timestamp)")
    @Transaction
    int[] bulkInsert(@Bind("timestamp") List<Timestamp> timestamp);

    @SqlQuery("SELECT timestamp FROM test")
    List<Timestamp> findAllTimestamps();

    @SqlUpdate("DELETE FROM test")
    boolean deleteAllTimestamps();

}

main executes this method:

public ScheduledExecutorService writeTimestampPeriodically(int consumerIntDelay, 

    int consumerPeriod) {
            buffer = new Buffer();
            TimestampService timestampService = new TimestampService();
            ScheduledExecutorService executorService = Executors.newScheduledThreadPool(3);
            executorService.scheduleAtFixedRate(new Producer("producer-1", buffer), 0, 1, TimeUnit.SECONDS);
            executorService.scheduleAtFixedRate(new Consumer("consumer-1", buffer, timestampService), consumerIntDelay, consumerPeriod, TimeUnit.SECONDS);
            executorService.scheduleAtFixedRate(new Monitor(buffer, "monitor-1"), 0, 1, TimeUnit.SECONDS);
            return executorService;
        }
\$\endgroup\$
  • 1
    \$\begingroup\$ The title and the description talk about a producer-consumer system, but the code you have presented for review is just a bunch of connections. Either post your solution (a GitHub link is not sufficient, as per the help center rules), or edit the question to describe what this code actually accomplishes. \$\endgroup\$ – 200_success Feb 27 at 3:30

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.