4
\$\begingroup\$

I have a program where I need to implement asynchronous tasks (writing a directory to a file following any change to the directory). There is existing documentation within my organization for implementing tasks, which includes actions and actors. Most of it deals with distributed activity across multiple servers, which my application does not deal with, so I've really had to strip out much of the logic implemented in other applications.

Whenever an action is taken that modifies the directory, I create the appropriate action and an actor, and then add the actor to the list of actors (as long as the actor does not already exist). I then execute the synchronous logic in the action object, and the asynchronous logic in the actor object. There is a 3 second sleep added for mimicking asynchronous operation as this program is more of an example for testing.

What I mostly would like input on is my synchronization and processing of async tasks. I am trying to maintain consistency by ensuring the file is not written to while another write operation is happening, but I also copied over the synchronization when creating actors from existing code. Is this a good approach?

public class UserDirectoryActor {

private final Action<UserEntry> action;
private final ExecutorService pool = Executors.newFixedThreadPool(1);

public UserDirectoryActor(Action<UserEntry> action) {
    this.action = action;
}

public Action<UserEntry> getAction() {
    return action;
}

/*
 * To simulate asynchronous processing, there is a 3 second wait
 * between the start of the asynchronous operation and the
 * actual writing of the user directory back to the file
 */
public Future<String> executeAsynchronously() {
    UserServiceFw.log.debug("Writing directory to file...");
    Callable<String> asyncTask = () -> {
        UserServiceFw.entryManager.writeBack();
        return "Finished!";
    };
    try {
        action.execute();
        TimeUnit.SECONDS.sleep(3);
        UserServiceFw.log.debug("Finished writing directory to file!");
    } catch (InterruptedException e) {
        UserServiceFw.log.error("Asynchronous wait interrupted by exception: " + e.toString());
    }
    return pool.submit(asyncTask);
}
}

Here is one of the action classes (the business logic of the actions is the only thing that differs):

public class AddEntryAction {

public AddEntryAction() {
}

/*
 * @return true if entry successfully added
 */
public void execute(UserEntry entryModel, boolean action) {
    if (!UserServiceFw.entryManager.directory.contains(entryModel)) {
        if (entryModel.getFirstName() != null && entryModel.getLastName() != null && entryModel.getAddress() != null &&
                entryModel.getPhoneNumber() != null) {
            UserServiceFw.entryManager.directory.add(entryModel);
        }
    }
}
}

The actions are used in this class:

public class UserEntryManager {

public final ArrayList<UserEntry> directory = new ArrayList<>();
private static File f;
private final Set<UserDirectoryActor> actors;

public UserEntryManager() throws BusinessException {
    ...
    actors = new CopyOnWriteArraySet<>();
}

private void readIn() throws BusinessException {
    ...
}

private synchronized void writeOut() throws BusinessException {
    if(f.exists() && f.canWrite()) {
        try (FileWriter fileWriter = new FileWriter(f)) {
            String endLine = "\n";
            fileWriter.write("");
            for(UserEntry entry : directory) {
                fileWriter.append(entry.getLastName());
                fileWriter.append(CSV_DELIMITER);
                fileWriter.append(entry.getFirstName());
                fileWriter.append(CSV_DELIMITER);
                fileWriter.append(entry.getPhoneNumber());
                fileWriter.append(CSV_DELIMITER);
                fileWriter.append(entry.getAddress());
                fileWriter.append(endLine);
            }
        } catch (IOException e) {
            UserServiceFw.log.error("Error creating FileWriter: " + e.getMessage());
            throw new BusinessException(e);
        }
    } else {
        UserServiceFw.log.error(f.getAbsolutePath() + " doesn't exist or can't be written to");
    }
}

public void addEntry(UserkEntry entryModel, boolean notify) {
    assert entryModel != null;
    synchronized (this) {
        AddEntryAction action = new AddEntryAction();
        UserDirectoryActor actor = new UserDirectoryActor(action);
        addActor(actor);
        actor.executeAsynchronously();
        if (notify) {
             //Send notification
        }
    removeActor(actor);
    }
}

public void deleteEntry(UserEntry entryModel, boolean notify) {
    assert entryModel != null;
    synchronized (this) {
        DeleteEntryAction action = new DeleteEntryAction();
        UserDirectoryActor actor = new UserDirectoryActor(action);
        addActor(actor);
        actor.executeAsynchronously();
        if(notify) {
            //Send notification
        }
    removeActor(actor);
    }
}

public void updateEntry(final UserEntry oldEntry, final UserEntry newEntry, boolean notify) {
    assert oldEntry != null && newEntry != null;
    synchronized (this) {
        UpdateEntryAction action = new UpdateEntryAction();
        UserDirectoryActor actor = new UserDirectoryActor(action);
        addActor(actor);
        actor.executeAsynchronously();
        if(notify) {
           //Send notification
        }
    removeActor(actor);
    }
}

protected final synchronized void addActor(UserDirectoryActor actor) throws BusinessException {
    for(UserDirectoryActor act : actors) {
        if(act.equals(actor)) {
            UserServiceFw.log.error("Actor for " + act.getAction() + " already exists");
            throw new BusinessException("UserDirectoryActor could not be added for: " + act.getAction());
        }
    }
    actors.add(actor);
}

protected final synchronized boolean deleteActor(UserDirectoryActor actor) {
    for(UserDirectoryActor act : actors) {
        if(act.equals(actor)) {
            actors.remove(act);
            return true;
        }
    }
    return false;
}

public ArrayList<UserEntry> getEntry(String firstName, String lastName) {
    assert firstName != null && lastName != null;
    return directory.stream().filter(entry -> entry.getFirstName().compareTo(firstName) == 0 &&
            entry.getLastName().compareTo(lastName) == 0).collect(Collectors.toCollection(ArrayList::new));
}

public void writeBack() throws BusinessException {
    Collections.sort(directory);
    writeOut();
}
}
\$\endgroup\$

1 Answer 1

2
\$\begingroup\$

I would reshape your code in a way that any actor is an object that has a blocking queue as an inbox, a queue as an outbox and main loop (the run() method) that just queries the inbox, processes its stuff and writes to the outbox ad infinitum. Then you'd have a class that does the wiring for you: It would create all the actors, connect them through SynchronizedBlockingQueues and then add all actors to a thread pool of the correct size.

An interesting alternative to Actors could be using a Disruptor. This is a framework for throwing data around between threads and managing parallel processing and batching, which from the programmer's perspective, feels quite similar to actors. Their web site has pretty good tutorials and you'll find it easy to work with. The basic concept of a Disruptor is a ring buffer that has slots for data/events, there can only one writer to the ring buffer, but multiple readers (you might like to call them "event processors"). They are organized in a way that they can follow (and race) each other, but are guaranteed to to the jobs in the defined order. The ring buffer is sort of a common queue, but it's a bit more sophisticated than that. Pretty ingenious stuff and a lot of work and analysis went into this. It was used to implement the asynchronous loggers in Log4j2.

You might like to read my blog post about actors, where you can see an example of an actor model, implemented in plain java, including the pitfalls.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.